Team Ai
Datasetpublic

codekingpro/portable-devtools

sourceHugging Faceupdated 5mo agoView on Hugging Face
1likes14kdownloads
_raw_display_base.py923 linesDownload Raw Back to display
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 abc28import contextlib29import functools30import os31import platform32import selectors33import signal34import socket35import sys36import typing37 38from urwid import signals, str_util, util39 40from . import escape41from .common import UNPRINTABLE_TRANS_TABLE, UPDATE_PALETTE_ENTRY, AttrSpec, BaseScreen, RealTerminal42 43if typing.TYPE_CHECKING:44    from collections.abc import Callable, Iterable45    from types import FrameType46 47    from typing_extensions import Literal48 49    from urwid import Canvas, EventLoop50 51IS_WINDOWS = sys.platform == "win32"52IS_WSL = (sys.platform == "linux") and ("wsl" in platform.platform().lower())53 54 55class Screen(BaseScreen, RealTerminal):56    def __init__(self, input: typing.IO, output: typing.IO) -> None:  # noqa: A002  # pylint: disable=redefined-builtin57        """Initialize a screen that directly prints escape codes to an output58        terminal.59        """60        super().__init__()61 62        self._partial_codes: list[int] = []63        self._pal_escape: dict[str | None, str] = {}64        self._pal_attrspec: dict[str | None, AttrSpec] = {}65        self._alternate_buffer: bool = False66        signals.connect_signal(self, UPDATE_PALETTE_ENTRY, self._on_update_palette_entry)67        self.colors: Literal[1, 16, 88, 256, 16777216] = 16  # FIXME: detect this68        self.has_underline = True  # FIXME: detect this69        self.prev_input_resize = 070        self.set_input_timeouts()71        self.screen_buf = None72        self._screen_buf_canvas = None73        self._resized = False74        self.maxrow = None75        self._mouse_tracking_enabled = False76        self.last_bstate = 077        self._setup_G1_done = False78        self._rows_used = None79        self._cy = 080        self.term = os.environ.get("TERM", "")81        self.fg_bright_is_bold = not self.term.startswith("xterm")82        self.bg_bright_is_blink = self.term == "linux"83        self.back_color_erase = not self.term.startswith("screen")84        self.register_palette_entry(None, "default", "default")85        self._next_timeout = None86        self.signal_handler_setter = signal.signal87 88        # Our connections to the world89        self._term_output_file = output90        self._term_input_file = input91 92        # pipe for signalling external event loops about resize events93        self._resize_pipe_rd, self._resize_pipe_wr = socket.socketpair()94        self._resize_pipe_rd.setblocking(False)95 96    def __del__(self) -> None:97        self._resize_pipe_rd.close()98        self._resize_pipe_wr.close()99 100    def __repr__(self) -> str:101        return f"<{self.__class__.__name__}(input={self._term_input_file}, output={self._term_output_file})>"102 103    def _sigwinch_handler(self, signum: int = 28, frame: FrameType | None = None) -> None:104        """105        frame -- will always be None when the GLib event loop is being used.106        """107        logger = self.logger.getChild("signal_handlers")108 109        logger.debug(f"SIGWINCH handler called with signum={signum!r}, frame={frame!r}")110 111        if IS_WINDOWS or not self._resized:112            self._resize_pipe_wr.send(b"R")113            logger.debug("Sent fake resize input to the pipe")114        self._resized = True115        self.screen_buf = None116 117    @property118    def _term_input_io(self) -> typing.IO | None:119        if hasattr(self._term_input_file, "fileno"):120            return self._term_input_file121        return None122 123    def _input_fileno(self) -> int | None:124        """Returns the fileno of the input stream, or None if it doesn't have one.125 126        A stream without a fileno can't participate in whatever.127        """128        if hasattr(self._term_input_file, "fileno"):129            return self._term_input_file.fileno()130 131        return None132 133    def _on_update_palette_entry(self, name: str | None, *attrspecs: AttrSpec):134        # copy the attribute to a dictionary containing the escape seqences135        a: AttrSpec = attrspecs[{16: 0, 1: 1, 88: 2, 256: 3, 2**24: 4}[self.colors]]136        self._pal_attrspec[name] = a137        self._pal_escape[name] = self._attrspec_to_escape(a)138 139    def set_input_timeouts(140        self,141        max_wait: float | None = None,142        complete_wait: float = 0.125,143        resize_wait: float = 0.125,144    ) -> None:145        """146        Set the get_input timeout values.  All values are in floating147        point numbers of seconds.148 149        max_wait -- amount of time in seconds to wait for input when150            there is no input pending, wait forever if None151        complete_wait -- amount of time in seconds to wait when152            get_input detects an incomplete escape sequence at the153            end of the available input154        resize_wait -- amount of time in seconds to wait for more input155            after receiving two screen resize requests in a row to156            stop Urwid from consuming 100% cpu during a gradual157            window resize operation158        """159        self.max_wait = max_wait160        if max_wait is not None:161            if self._next_timeout is None:162                self._next_timeout = max_wait163            else:164                self._next_timeout = min(self._next_timeout, self.max_wait)165        self.complete_wait = complete_wait166        self.resize_wait = resize_wait167 168    def set_mouse_tracking(self, enable: bool = True) -> None:169        """170        Enable (or disable) mouse tracking.171 172        After calling this function get_input will include mouse173        click events along with keystrokes.174        """175        enable = bool(enable)176        if enable == self._mouse_tracking_enabled:177            return178 179        self._mouse_tracking(enable)180        self._mouse_tracking_enabled = enable181 182    def _mouse_tracking(self, enable: bool) -> None:183        if enable:184            self.write(escape.MOUSE_TRACKING_ON)185        else:186            self.write(escape.MOUSE_TRACKING_OFF)187 188    @abc.abstractmethod189    def _start(self, alternate_buffer: bool = True) -> None:190        """191        Initialize the screen and input mode.192 193        alternate_buffer -- use alternate screen buffer194        """195 196    def _stop_mouse_restore_buffer(self) -> None:197        """Stop mouse tracking and restore the screen."""198        self._mouse_tracking(False)199 200        move_cursor = ""201        if self._alternate_buffer:202            move_cursor = escape.RESTORE_NORMAL_BUFFER203        elif self.maxrow is not None:204            move_cursor = escape.set_cursor_position(0, self.maxrow)205        self.write(self._attrspec_to_escape(AttrSpec("", "")) + escape.SI + move_cursor + escape.SHOW_CURSOR)206        self.flush()207 208    @abc.abstractmethod209    def _stop(self) -> None:210        """211        Restore the screen.212        """213 214    def write(self, data: str) -> None:215        """Write some data to the terminal.216 217        You may wish to override this if you're using something other than218        regular files for input and output.219        """220        self._term_output_file.write(data)221 222    def flush(self) -> None:223        """Flush the output buffer.224 225        You may wish to override this if you're using something other than226        regular files for input and output.227        """228        self._term_output_file.flush()229 230    @typing.overload231    def get_input(self, raw_keys: Literal[False]) -> list[str]: ...232 233    @typing.overload234    def get_input(self, raw_keys: Literal[True]) -> tuple[list[str], list[int]]: ...235 236    def get_input(self, raw_keys: bool = False) -> list[str] | tuple[list[str], list[int]]:237        """Return pending input as a list.238 239        raw_keys -- return raw keycodes as well as translated versions240 241        This function will immediately return all the input since the242        last time it was called.  If there is no input pending it will243        wait before returning an empty list.  The wait time may be244        configured with the set_input_timeouts function.245 246        If raw_keys is False (default) this function will return a list247        of keys pressed.  If raw_keys is True this function will return248        a ( keys pressed, raw keycodes ) tuple instead.249 250        Examples of keys returned:251 252        * ASCII printable characters:  " ", "a", "0", "A", "-", "/"253        * ASCII control characters:  "tab", "enter"254        * Escape sequences:  "up", "page up", "home", "insert", "f1"255        * Key combinations:  "shift f1", "meta a", "ctrl b"256        * Window events:  "window resize"257 258        When a narrow encoding is not enabled:259 260        * "Extended ASCII" characters:  "\\xa1", "\\xb2", "\\xfe"261 262        When a wide encoding is enabled:263 264        * Double-byte characters:  "\\xa1\\xea", "\\xb2\\xd4"265 266        When utf8 encoding is enabled:267 268        * Unicode characters: u"\\u00a5", u'\\u253c"269 270        Examples of mouse events returned:271 272        * Mouse button press: ('mouse press', 1, 15, 13),273                              ('meta mouse press', 2, 17, 23)274        * Mouse drag: ('mouse drag', 1, 16, 13),275                      ('mouse drag', 1, 17, 13),276                      ('ctrl mouse drag', 1, 18, 13)277        * Mouse button release: ('mouse release', 0, 18, 13),278                                ('ctrl mouse release', 0, 17, 23)279        """280        logger = self.logger.getChild("get_input")281        if not self._started:282            raise RuntimeError283 284        self._wait_for_input_ready(self._next_timeout)285        keys, raw = self.parse_input(None, None, self.get_available_raw_input())286 287        # Avoid pegging CPU at 100% when slowly resizing288        if keys == ["window resize"] and self.prev_input_resize:289            logger.debug('get_input: got "window resize" > 1 times. Enable throttling for resize.')290            for _ in range(2):291                self._wait_for_input_ready(self.resize_wait)292                new_keys, new_raw = self.parse_input(None, None, self.get_available_raw_input())293                raw += new_raw294                if new_keys and new_keys != ["window resize"]:295                    if "window resize" in new_keys:296                        keys = new_keys297                    else:298                        keys.extend(new_keys)299                    break300 301        if keys == ["window resize"]:302            self.prev_input_resize = 2303        elif self.prev_input_resize == 2 and not keys:304            self.prev_input_resize = 1305        else:306            self.prev_input_resize = 0307 308        if raw_keys:309            return keys, raw310        return keys311 312    def get_input_descriptors(self) -> list[socket.socket | typing.IO | int]:313        """314        Return a list of integer file descriptors that should be315        polled in external event loops to check for user input.316 317        Use this method if you are implementing your own event loop.318 319        This method is only called by `hook_event_loop`, so if you override320        that, you can safely ignore this.321        """322        if not self._started:323            return []324 325        fd_list: list[socket.socket | typing.IO | int] = [self._resize_pipe_rd]326 327        if (input_io := self._term_input_io) is not None:328            fd_list.append(input_io)329        return fd_list330 331    _current_event_loop_handles = ()332 333    @abc.abstractmethod334    def unhook_event_loop(self, event_loop: EventLoop) -> None:335        """336        Remove any hooks added by hook_event_loop.337        """338 339    @abc.abstractmethod340    def hook_event_loop(341        self,342        event_loop: EventLoop,343        callback: Callable[[list[str], list[int]], typing.Any],344    ) -> None:345        """346        Register the given callback with the event loop, to be called with new347        input whenever it's available.  The callback should be passed a list of348        processed keys and a list of unprocessed keycodes.349 350        Subclasses may wish to use parse_input to wrap the callback.351        """352 353    _input_timeout = None354 355    def _make_legacy_input_wrapper(356        self,357        event_loop: EventLoop,358        callback: Callable[[list[str], list[int]], typing.Any],359    ) -> Callable[[], None]:360        """361        Support old Screen classes that still have a get_input_nonblocking and expect it to work.362        """363 364        @functools.wraps(callback)365        def wrapper():366            if self._input_timeout:367                event_loop.remove_alarm(self._input_timeout)368                self._input_timeout = None369            timeout, keys, raw = self.get_input_nonblocking()  # pylint: disable=no-member  # should we deprecate?370            if timeout is not None:371                self._input_timeout = event_loop.alarm(timeout, wrapper)372 373            callback(keys, raw)374 375        return wrapper376 377    def _get_input_codes(self) -> list[int]:378        return list(self._get_keyboard_codes())379 380    def get_available_raw_input(self) -> list[int]:381        """382        Return any currently available input. Does not block.383 384        This method is only used by the default `hook_event_loop`385        implementation; you can safely ignore it if you implement your own.386        """387        logger = self.logger.getChild("get_available_raw_input")388        codes = [*self._partial_codes, *self._get_input_codes()]389        self._partial_codes = []390 391        # clean out the pipe used to signal external event loops392        # that a resize has occurred393        with selectors.DefaultSelector() as selector:394            selector.register(self._resize_pipe_rd, selectors.EVENT_READ)395            present_resize_flag = selector.select(0)  # nonblocking396            while present_resize_flag:397                logger.debug("Resize signal received. Cleaning socket.")398                # Argument "size" is maximum buffer size to read. Since we're emptying, set it reasonably big.399                self._resize_pipe_rd.recv(128)400                present_resize_flag = selector.select(0)401 402        return codes403 404    @typing.overload405    def parse_input(406        self,407        event_loop: None,408        callback: None,409        codes: list[int],410        wait_for_more: bool = ...,411    ) -> tuple[list[str], list[int]]: ...412 413    @typing.overload414    def parse_input(415        self,416        event_loop: EventLoop,417        callback: None,418        codes: list[int],419        wait_for_more: bool = ...,420    ) -> tuple[list[str], list[int]]: ...421 422    @typing.overload423    def parse_input(424        self,425        event_loop: EventLoop,426        callback: Callable[[list[str], list[int]], typing.Any],427        codes: list[int],428        wait_for_more: bool = ...,429    ) -> None: ...430 431    def parse_input(432        self,433        event_loop: EventLoop | None,434        callback: Callable[[list[str], list[int]], typing.Any] | None,435        codes: list[int],436        wait_for_more: bool = True,437    ) -> tuple[list[str], list[int]] | None:438        """439        Read any available input from get_available_raw_input, parses it into440        keys, and calls the given callback.441 442        The current implementation tries to avoid any assumptions about what443        the screen or event loop look like; it only deals with parsing keycodes444        and setting a timeout when an incomplete one is detected.445 446        `codes` should be a sequence of keycodes, i.e. bytes.  A bytearray is447        appropriate, but beware of using bytes, which only iterates as integers448        on Python 3.449        """450 451        logger = self.logger.getChild("parse_input")452 453        # Note: event_loop may be None for 100% synchronous support, only used454        # by get_input.  Not documented because you shouldn't be doing it.455        if self._input_timeout and event_loop:456            event_loop.remove_alarm(self._input_timeout)457            self._input_timeout = None458 459        original_codes = codes460        decoded_codes = []461        try:462            while codes:463                run, codes = escape.process_keyqueue(codes, wait_for_more)464                decoded_codes.extend(run)465        except escape.MoreInputRequired:466            # Set a timer to wait for the rest of the input; if it goes off467            # without any new input having come in, use the partial input468            k = len(original_codes) - len(codes)469            raw_codes = original_codes[:k]470            self._partial_codes = codes471 472            def _parse_incomplete_input():473                self._input_timeout = None474                self._partial_codes = []475                self.parse_input(event_loop, callback, codes, wait_for_more=False)476 477            if event_loop:478                self._input_timeout = event_loop.alarm(self.complete_wait, _parse_incomplete_input)479 480        else:481            raw_codes = original_codes482            self._partial_codes = []483 484        logger.debug(f"Decoded codes: {decoded_codes!r}, raw codes: {raw_codes!r}")485 486        if self._resized:487            decoded_codes.append("window resize")488            logger.debug('Added "window resize" to the codes')489            self._resized = False490 491        if callback:492            callback(decoded_codes, raw_codes)493            return None494 495        # For get_input496        return decoded_codes, raw_codes497 498    def _wait_for_input_ready(self, timeout: float | None) -> list[int]:499        logger = self.logger.getChild("wait_for_input_ready")500        fd_list = self.get_input_descriptors()501 502        logger.debug(f"Waiting for input: descriptors={fd_list!r}, timeout={timeout!r}")503        with selectors.DefaultSelector() as selector:504            for fd in fd_list:505                selector.register(fd, selectors.EVENT_READ)506 507            ready = selector.select(timeout)508 509        logger.debug(f"Input ready: {ready}")510 511        return [event.fd for event, _ in ready]512 513    @abc.abstractmethod514    def _read_raw_input(self, timeout: int) -> Iterable[int]: ...515 516    def _get_keyboard_codes(self) -> Iterable[int]:517        return self._read_raw_input(0)518 519    def _setup_G1(self) -> None:520        """521        Initialize the G1 character set to graphics mode if required.522        """523        if self._setup_G1_done:524            return525 526        while True:527            with contextlib.suppress(OSError):528                self.write(escape.DESIGNATE_G1_SPECIAL)529                self.flush()530                break531        self._setup_G1_done = True532 533    def draw_screen(self, size: tuple[int, int], canvas: Canvas) -> None:534        """Paint screen with rendered canvas."""535 536        def set_cursor_home() -> str:537            if not partial_display():538                return escape.set_cursor_position(0, 0)539            return escape.CURSOR_HOME_COL + escape.move_cursor_up(cy)540 541        def set_cursor_position(x: int, y: int) -> str:542            if not partial_display():543                return escape.set_cursor_position(x, y)544            if cy > y:545                return "\b" + escape.CURSOR_HOME_COL + escape.move_cursor_up(cy - y) + escape.move_cursor_right(x)546            return "\b" + escape.CURSOR_HOME_COL + escape.move_cursor_down(y - cy) + escape.move_cursor_right(x)547 548        def is_blank_row(row: list[tuple[object, Literal["0", "U"] | None], bytes]) -> bool:549            if len(row) > 1:550                return False551            return not row[0][2].strip()552 553        def attr_to_escape(a: AttrSpec | str | None) -> str:554            if a in self._pal_escape:555                return self._pal_escape[a]556            if isinstance(a, AttrSpec):557                return self._attrspec_to_escape(a)558            if a is None:559                return self._attrspec_to_escape(AttrSpec("default", "default"))560            # undefined attributes use default/default561            self.logger.debug(f"Undefined attribute: {a!r}")562            return self._attrspec_to_escape(AttrSpec("default", "default"))563 564        def using_standout_or_underline(a: AttrSpec | str) -> bool:565            a = self._pal_attrspec.get(a, a)566            return isinstance(a, AttrSpec) and (a.standout or a.underline)567 568        encoding = util.get_encoding()569 570        logger = self.logger.getChild("draw_screen")571 572        (maxcol, maxrow) = size573 574        if not self._started:575            raise RuntimeError576 577        if maxrow != canvas.rows():578            raise ValueError(maxrow)579 580        # quick return if nothing has changed581        if self.screen_buf and canvas is self._screen_buf_canvas:582            return583 584        self._setup_G1()585 586        if self._resized:587            # handle resize before trying to draw screen588            logger.debug("Not drawing screen: screen resized and resize was not handled")589            return590 591        logger.debug(f"Drawing screen with size {size!r}")592 593        last_attributes = None  # Default = empty594 595        output: list[str] = [escape.HIDE_CURSOR, attr_to_escape(last_attributes)]596 597        def partial_display() -> bool:598            # returns True if the screen is in partial display mode ie. only some rows belong to the display599            return self._rows_used is not None600 601        if not partial_display():602            output.append(escape.CURSOR_HOME)603 604        if self.screen_buf:605            osb = self.screen_buf606        else:607            osb = []608        sb: list[list[tuple[object, Literal["0", "U"] | None, bytes]]] = []609        cy = self._cy610        y = -1611 612        ins = None613        output.append(set_cursor_home())614        cy = 0615 616        first = True617        last_charset_flag = None618 619        for row in canvas.content():620            y += 1621            if osb and y < len(osb) and osb[y] == row:622                # this row of the screen buffer matches what is623                # currently displayed, so we can skip this line624                sb.append(osb[y])625                continue626 627            sb.append(row)628 629            # leave blank lines off display when we are using630            # the default screen buffer (allows partial screen)631            if partial_display() and y > self._rows_used:632                if is_blank_row(row):633                    continue634                self._rows_used = y635 636            if y or partial_display():637                output.append(set_cursor_position(0, y))638            # after updating the line we will be just over the639            # edge, but terminals still treat this as being640            # on the same line641            cy = y642 643            whitespace_at_end = False644            if row:645                a, cs, run = row[-1]646                if run[-1:] == b" " and self.back_color_erase and not using_standout_or_underline(a):647                    whitespace_at_end = True648                    row = [*row[:-1], (a, cs, run.rstrip(b" "))]  # noqa: PLW2901649                elif y == maxrow - 1 and maxcol > 1:650                    row, back, ins = self._last_row(row)  # noqa: PLW2901651 652            for a, cs, run in row:653                if not isinstance(run, bytes):  # canvases render with bytes654                    raise TypeError(run)655 656                if cs != "U":657                    run = run.translate(UNPRINTABLE_TRANS_TABLE)  # noqa: PLW2901658 659                if last_attributes != a:660                    output.append(attr_to_escape(a))661                    last_attributes = a662 663                if encoding != "utf-8" and (first or last_charset_flag != cs):664                    if cs not in {None, "0", "U"}:665                        raise ValueError(cs)666                    if last_charset_flag == "U":667                        output.append(escape.IBMPC_OFF)668 669                    if cs is None:670                        output.append(escape.SI)671                    elif cs == "U":672                        output.append(escape.IBMPC_ON)673                    else:674                        output.append(escape.SO)675                    last_charset_flag = cs676 677                output.append(run.decode(encoding, "replace"))678                first = False679 680            if ins:681                (inserta, insertcs, inserttext) = ins682                ias = attr_to_escape(inserta)683                if insertcs not in {None, "0", "U"}:684                    raise ValueError(insertcs)685 686                if isinstance(inserttext, bytes):687                    inserttext = inserttext.decode(encoding)688 689                output.extend(("\x08" * back, ias))  # pylint: disable=used-before-assignment  # defined in `if row`690 691                if encoding != "utf-8":692                    if cs is None:693                        icss = escape.SI694                    elif cs == "U":695                        icss = escape.IBMPC_ON696                    else:697                        icss = escape.SO698 699                    output.append(icss)700 701                if not IS_WINDOWS:702                    output += [escape.INSERT_ON, inserttext, escape.INSERT_OFF]703                else:704                    output += [f"{escape.ESC}[{str_util.calc_width(inserttext, 0, len(inserttext))}@", inserttext]705 706                if encoding != "utf-8" and cs == "U":707                    output.append(escape.IBMPC_OFF)708 709            if whitespace_at_end:710                output.append(escape.ERASE_IN_LINE_RIGHT)711 712        if canvas.cursor is not None:713            x, y = canvas.cursor714            output += [set_cursor_position(x, y), escape.SHOW_CURSOR]715            self._cy = y716 717        if self._resized:718            # handle resize before trying to draw screen719            return720        try:721            for line in output:722                if isinstance(line, bytes):723                    line = line.decode(encoding, "replace")  # noqa: PLW2901724                self.write(line)725            self.flush()726        except OSError as e:727            # ignore interrupted syscall728            if e.args[0] != 4:729                raise730 731        self.screen_buf = sb732        self._screen_buf_canvas = canvas733 734    def _last_row(735        self,736        row: list[tuple[object, Literal["0", "U"] | None, bytes]],737    ) -> tuple[738        list[tuple[object, Literal["0", "U"] | None, bytes]],739        int,740        tuple[object, Literal["0", "U"] | None, bytes],741    ]:742        """On the last row we need to slide the bottom right character743        into place. Calculate the new line, attr and an insert sequence744        to do that.745 746        eg. last row:747        XXXXXXXXXXXXXXXXXXXXYZ748 749        Y will be drawn after Z, shifting Z into position.750        """751 752        new_row = row[:-1]753        z_attr, z_cs, last_text = row[-1]754        last_cols = str_util.calc_width(last_text, 0, len(last_text))755        last_offs, z_col = str_util.calc_text_pos(last_text, 0, len(last_text), last_cols - 1)756        if last_offs == 0:757            z_text = last_text758            del new_row[-1]759            # we need another segment760            y_attr, y_cs, nlast_text = row[-2]761            nlast_cols = str_util.calc_width(nlast_text, 0, len(nlast_text))762            z_col += nlast_cols763            nlast_offs, y_col = str_util.calc_text_pos(nlast_text, 0, len(nlast_text), nlast_cols - 1)764            y_text = nlast_text[nlast_offs:]765            if nlast_offs:766                new_row.append((y_attr, y_cs, nlast_text[:nlast_offs]))767        else:768            z_text = last_text[last_offs:]769            y_attr, y_cs = z_attr, z_cs770            nlast_cols = str_util.calc_width(last_text, 0, last_offs)771            nlast_offs, y_col = str_util.calc_text_pos(last_text, 0, last_offs, nlast_cols - 1)772            y_text = last_text[nlast_offs:last_offs]773            if nlast_offs:774                new_row.append((y_attr, y_cs, last_text[:nlast_offs]))775 776        new_row.append((z_attr, z_cs, z_text))777        return new_row, z_col - y_col, (y_attr, y_cs, y_text)778 779    def clear(self) -> None:780        """781        Force the screen to be completely repainted on the next782        call to draw_screen().783        """784        self.screen_buf = None785 786    def _attrspec_to_escape(self, a: AttrSpec) -> str:787        """788        Convert AttrSpec instance a to an escape sequence for the terminal789 790        >>> s = Screen()791        >>> s.set_terminal_properties(colors=256)792        >>> a2e = s._attrspec_to_escape793        >>> a2e(s.AttrSpec("brown", "dark green"))794        '\\x1b[0;33;42m'795        >>> a2e(s.AttrSpec("#fea,underline", "#d0d"))796        '\\x1b[0;38;5;229;4;48;5;164m'797        """798        if self.term == "fbterm":799            fg = escape.ESC + f"[1;{a.foreground_number:d}}}"800            bg = escape.ESC + f"[2;{a.background_number:d}}}"801            return fg + bg802 803        if a.foreground_true:804            fg = f"38;2;{';'.join(str(part) for part in a.get_rgb_values()[0:3])}"805        elif a.foreground_high:806            fg = f"38;5;{a.foreground_number:d}"807        elif a.foreground_basic:808            if a.foreground_number > 7:809                if self.fg_bright_is_bold:810                    fg = f"1;{a.foreground_number - 8 + 30:d}"811                else:812                    fg = f"{a.foreground_number - 8 + 90:d}"813            else:814                fg = f"{a.foreground_number + 30:d}"815        else:816            fg = "39"817        st = (818            "1;" * a.bold819            + "3;" * a.italics820            + "4;" * a.underline821            + "5;" * a.blink822            + "7;" * a.standout823            + "9;" * a.strikethrough824        )825        if a.background_true:826            bg = f"48;2;{';'.join(str(part) for part in a.get_rgb_values()[3:6])}"827        elif a.background_high:828            bg = f"48;5;{a.background_number:d}"829        elif a.background_basic:830            if a.background_number > 7:831                if self.bg_bright_is_blink:832                    bg = f"5;{a.background_number - 8 + 40:d}"833                else:834                    # this doesn't work on most terminals835                    bg = f"{a.background_number - 8 + 100:d}"836            else:837                bg = f"{a.background_number + 40:d}"838        else:839            bg = "49"840        return f"{escape.ESC}[0;{fg};{st}{bg}m"841 842    def set_terminal_properties(843        self,844        colors: Literal[1, 16, 88, 256, 16777216] | None = None,845        bright_is_bold: bool | None = None,846        has_underline: bool | None = None,847    ) -> None:848        """849        colors -- number of colors terminal supports (1, 16, 88, 256, or 2**24)850            or None to leave unchanged851        bright_is_bold -- set to True if this terminal uses the bold852            setting to create bright colors (numbers 8-15), set to False853            if this Terminal can create bright colors without bold or854            None to leave unchanged855        has_underline -- set to True if this terminal can use the856            underline setting, False if it cannot or None to leave857            unchanged858        """859        if colors is None:860            colors = self.colors861        if bright_is_bold is None:862            bright_is_bold = self.fg_bright_is_bold863        if has_underline is None:864            has_underline = self.has_underline865 866        if colors == self.colors and bright_is_bold == self.fg_bright_is_bold and has_underline == self.has_underline:867            return868 869        self.colors = colors870        self.fg_bright_is_bold = bright_is_bold871        self.has_underline = has_underline872 873        self.clear()874        self._pal_escape = {}875        for p, v in self._palette.items():876            self._on_update_palette_entry(p, *v)877 878    def reset_default_terminal_palette(self) -> None:879        """880        Attempt to set the terminal palette to default values as taken from xterm.881        Uses a number of colors from the current set_terminal_properties() screen setting.882        """883        if self.colors == 1:884            return885        if self.colors == 2**24:886            colors = 256887        else:888            colors = self.colors889 890        def rgb_values(n) -> tuple[int | None, int | None, int | None]:891            if colors == 16:892                aspec = AttrSpec(f"h{n:d}", "", 256)893            else:894                aspec = AttrSpec(f"h{n:d}", "", colors)895            return aspec.get_rgb_values()[:3]896 897        entries = [(n, *rgb_values(n)) for n in range(min(colors, 256))]898        self.modify_terminal_palette(entries)899 900    def modify_terminal_palette(self, entries: list[tuple[int, int | None, int | None, int | None]]):901        """902        entries - list of (index, red, green, blue) tuples.903 904        Attempt to set part of the terminal palette (this does not work on all terminals.)905        The changes are sent as a single escape sequence so they should all take effect at the same time.906 907        0 <= index < 256 (some terminals will only have 16 or 88 colors)908        0 <= red, green, blue < 256909        """910 911        if self.term == "fbterm":912            modify = [f"{index:d};{red:d};{green:d};{blue:d}" for index, red, green, blue in entries]913            self.write(f"\x1b[3;{';'.join(modify)}}}")914        else:915            modify = [f"{index:d};rgb:{red:02x}/{green:02x}/{blue:02x}" for index, red, green, blue in entries]916            self.write(f"\x1b]4;{';'.join(modify)}\x1b\\")917        self.flush()918 919    # shortcut for creating an AttrSpec with this screen object's920    # number of colors921    def AttrSpec(self, fg, bg) -> AttrSpec:922        return AttrSpec(fg, bg, self.colors)923 
codekingpro/portable-devtools · Team Ai