codekingpro/portable-devtools

1# This module is based on the excellent work by Adam Bartoš who2# provided a lot of what went into the implementation here in3# the discussion to issue1602 in the Python bug tracker.4#5# There are some general differences in regards to how this works6# compared to the original patches as we do not need to patch7# the entire interpreter but just work in our little world of8# echo and prompt.9from __future__ import annotations10 11import collections.abc as cabc12import io13import sys14import time15import typing as t16from ctypes import Array17from ctypes import byref18from ctypes import c_char19from ctypes import c_char_p20from ctypes import c_int21from ctypes import c_ssize_t22from ctypes import c_ulong23from ctypes import c_void_p24from ctypes import POINTER25from ctypes import py_object26from ctypes import Structure27from ctypes.wintypes import DWORD28from ctypes.wintypes import HANDLE29from ctypes.wintypes import LPCWSTR30from ctypes.wintypes import LPWSTR31 32from ._compat import _NonClosingTextIOWrapper33 34assert sys.platform == "win32"35import msvcrt  # noqa: E40236from ctypes import windll  # noqa: E40237from ctypes import WINFUNCTYPE  # noqa: E40238 39c_ssize_p = POINTER(c_ssize_t)40 41kernel32 = windll.kernel3242GetStdHandle = kernel32.GetStdHandle43ReadConsoleW = kernel32.ReadConsoleW44WriteConsoleW = kernel32.WriteConsoleW45GetConsoleMode = kernel32.GetConsoleMode46GetLastError = kernel32.GetLastError47GetCommandLineW = WINFUNCTYPE(LPWSTR)(("GetCommandLineW", windll.kernel32))48CommandLineToArgvW = WINFUNCTYPE(POINTER(LPWSTR), LPCWSTR, POINTER(c_int))(49    ("CommandLineToArgvW", windll.shell32)50)51LocalFree = WINFUNCTYPE(c_void_p, c_void_p)(("LocalFree", windll.kernel32))52 53STDIN_HANDLE = GetStdHandle(-10)54STDOUT_HANDLE = GetStdHandle(-11)55STDERR_HANDLE = GetStdHandle(-12)56 57PyBUF_SIMPLE = 058PyBUF_WRITABLE = 159 60ERROR_SUCCESS = 061ERROR_NOT_ENOUGH_MEMORY = 862ERROR_OPERATION_ABORTED = 99563 64STDIN_FILENO = 065STDOUT_FILENO = 166STDERR_FILENO = 267 68EOF = b"\x1a"69MAX_BYTES_WRITTEN = 3276770 71if t.TYPE_CHECKING:72    try:73        # Using `typing_extensions.Buffer` instead of `collections.abc`74        # on Windows for some reason does not have `Sized` implemented.75        from collections.abc import Buffer  # type: ignore76    except ImportError:77        from typing_extensions import Buffer78 79try:80    from ctypes import pythonapi81except ImportError:82    # On PyPy we cannot get buffers so our ability to operate here is83    # severely limited.84    get_buffer = None85else:86 87    class Py_buffer(Structure):88        _fields_ = [  # noqa: RUF01289            ("buf", c_void_p),90            ("obj", py_object),91            ("len", c_ssize_t),92            ("itemsize", c_ssize_t),93            ("readonly", c_int),94            ("ndim", c_int),95            ("format", c_char_p),96            ("shape", c_ssize_p),97            ("strides", c_ssize_p),98            ("suboffsets", c_ssize_p),99            ("internal", c_void_p),100        ]101 102    PyObject_GetBuffer = pythonapi.PyObject_GetBuffer103    PyBuffer_Release = pythonapi.PyBuffer_Release104 105    def get_buffer(obj: Buffer, writable: bool = False) -> Array[c_char]:106        buf = Py_buffer()107        flags: int = PyBUF_WRITABLE if writable else PyBUF_SIMPLE108        PyObject_GetBuffer(py_object(obj), byref(buf), flags)109 110        try:111            buffer_type = c_char * buf.len112            out: Array[c_char] = buffer_type.from_address(buf.buf)113            return out114        finally:115            PyBuffer_Release(byref(buf))116 117 118class _WindowsConsoleRawIOBase(io.RawIOBase):119    def __init__(self, handle: int | None) -> None:120        self.handle = handle121 122    def isatty(self) -> t.Literal[True]:123        super().isatty()124        return True125 126 127class _WindowsConsoleReader(_WindowsConsoleRawIOBase):128    def readable(self) -> t.Literal[True]:129        return True130 131    def readinto(self, b: Buffer) -> int:132        bytes_to_be_read = len(b)133        if not bytes_to_be_read:134            return 0135        elif bytes_to_be_read % 2:136            raise ValueError(137                "cannot read odd number of bytes from UTF-16-LE encoded console"138            )139 140        buffer = get_buffer(b, writable=True)141        code_units_to_be_read = bytes_to_be_read // 2142        code_units_read = c_ulong()143 144        rv = ReadConsoleW(145            HANDLE(self.handle),146            buffer,147            code_units_to_be_read,148            byref(code_units_read),149            None,150        )151        if GetLastError() == ERROR_OPERATION_ABORTED:152            # wait for KeyboardInterrupt153            time.sleep(0.1)154        if not rv:155            raise OSError(f"Windows error: {GetLastError()}")156 157        if buffer[0] == EOF:158            return 0159        return 2 * code_units_read.value160 161 162class _WindowsConsoleWriter(_WindowsConsoleRawIOBase):163    def writable(self) -> t.Literal[True]:164        return True165 166    @staticmethod167    def _get_error_message(errno: int) -> str:168        if errno == ERROR_SUCCESS:169            return "ERROR_SUCCESS"170        elif errno == ERROR_NOT_ENOUGH_MEMORY:171            return "ERROR_NOT_ENOUGH_MEMORY"172        return f"Windows error {errno}"173 174    def write(self, b: Buffer) -> int:175        bytes_to_be_written = len(b)176        buf = get_buffer(b)177        code_units_to_be_written = min(bytes_to_be_written, MAX_BYTES_WRITTEN) // 2178        code_units_written = c_ulong()179 180        WriteConsoleW(181            HANDLE(self.handle),182            buf,183            code_units_to_be_written,184            byref(code_units_written),185            None,186        )187        bytes_written = 2 * code_units_written.value188 189        if bytes_written == 0 and bytes_to_be_written > 0:190            raise OSError(self._get_error_message(GetLastError()))191        return bytes_written192 193 194class ConsoleStream:195    def __init__(self, text_stream: t.TextIO, byte_stream: t.BinaryIO) -> None:196        self._text_stream = text_stream197        self.buffer = byte_stream198 199    @property200    def name(self) -> str:201        return self.buffer.name202 203    def write(self, x: t.AnyStr) -> int:204        if isinstance(x, str):205            return self._text_stream.write(x)206        try:207            self.flush()208        except Exception:209            pass210        return self.buffer.write(x)211 212    def writelines(self, lines: cabc.Iterable[t.AnyStr]) -> None:213        for line in lines:214            self.write(line)215 216    def __getattr__(self, name: str) -> t.Any:217        return getattr(self._text_stream, name)218 219    def isatty(self) -> bool:220        return self.buffer.isatty()221 222    def __repr__(self) -> str:223        return f"<ConsoleStream name={self.name!r} encoding={self.encoding!r}>"224 225 226def _get_text_stdin(buffer_stream: t.BinaryIO) -> t.TextIO:227    text_stream = _NonClosingTextIOWrapper(228        io.BufferedReader(_WindowsConsoleReader(STDIN_HANDLE)),229        "utf-16-le",230        "strict",231        line_buffering=True,232    )233    return t.cast(t.TextIO, ConsoleStream(text_stream, buffer_stream))234 235 236def _get_text_stdout(buffer_stream: t.BinaryIO) -> t.TextIO:237    text_stream = _NonClosingTextIOWrapper(238        io.BufferedWriter(_WindowsConsoleWriter(STDOUT_HANDLE)),239        "utf-16-le",240        "strict",241        line_buffering=True,242    )243    return t.cast(t.TextIO, ConsoleStream(text_stream, buffer_stream))244 245 246def _get_text_stderr(buffer_stream: t.BinaryIO) -> t.TextIO:247    text_stream = _NonClosingTextIOWrapper(248        io.BufferedWriter(_WindowsConsoleWriter(STDERR_HANDLE)),249        "utf-16-le",250        "strict",251        line_buffering=True,252    )253    return t.cast(t.TextIO, ConsoleStream(text_stream, buffer_stream))254 255 256_stream_factories: cabc.Mapping[int, t.Callable[[t.BinaryIO], t.TextIO]] = {257    0: _get_text_stdin,258    1: _get_text_stdout,259    2: _get_text_stderr,260}261 262 263def _is_console(f: t.TextIO) -> bool:264    if not hasattr(f, "fileno"):265        return False266 267    try:268        fileno = f.fileno()269    except (OSError, io.UnsupportedOperation):270        return False271 272    handle = msvcrt.get_osfhandle(fileno)273    return bool(GetConsoleMode(handle, byref(DWORD())))274 275 276def _get_windows_console_stream(277    f: t.TextIO, encoding: str | None, errors: str | None278) -> t.TextIO | None:279    if (280        get_buffer is None281        or encoding not in {"utf-16-le", None}282        or errors not in {"strict", None}283        or not _is_console(f)284    ):285        return None286 287    func = _stream_factories.get(f.fileno())288    if func is None:289        return None290 291    b = getattr(f, "buffer", None)292 293    if b is None:294        return None295 296    return func(b)297