cpython/Lib/_pyrepl/unix_console.py

#   Copyright 2000-2010 Michael Hudson-Doyle <micahel@gmail.com>
#                       Antonio Cuni
#                       Armin Rigo
#
#                        All Rights Reserved
#
#
# Permission to use, copy, modify, and distribute this software and
# its documentation for any purpose is hereby granted without fee,
# provided that the above copyright notice appear in all copies and
# that both that copyright notice and this permission notice appear in
# supporting documentation.
#
# THE AUTHOR MICHAEL HUDSON DISCLAIMS ALL WARRANTIES WITH REGARD TO
# THIS SOFTWARE, INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY
# AND FITNESS, IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL,
# INDIRECT OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER
# RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF
# CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN
# CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.

from __future__ import annotations

import errno
import os
import re
import select
import signal
import struct
import termios
import time
import types
import platform
from collections.abc import Callable
from dataclasses import dataclass
from fcntl import ioctl
from typing import TYPE_CHECKING, cast, overload

from _colorize import ANSIColors

from . import terminfo
from .console import Console, Event
from .fancy_termios import tcgetattr, tcsetattr, TermState
from .render import (
    EMPTY_RENDER_LINE,
    LineUpdate,
    RenderLine,
    RenderedScreen,
    requires_cursor_resync,
    diff_render_lines,
    render_cells,
)
from .trace import trace, trace_text
from .unix_eventqueue import EventQueue

# declare posix optional to allow None assignment on other platforms
posix: types.ModuleType | None
try:
    import posix
except ImportError:
    posix = None

# types
if TYPE_CHECKING:
    from typing import AbstractSet, IO, Literal

type _MoveFunc = Callable[[int, int], None]
type _PendingWrite = tuple[str | bytes, bool]


class InvalidTerminal(RuntimeError):
    def __init__(self, message: str) -> None:
        super().__init__(errno.EIO, message)


_error = (termios.error, InvalidTerminal)
_error_codes_to_ignore = frozenset([errno.EIO, errno.ENXIO, errno.EPERM])

SIGWINCH_EVENT = "repaint"

FIONREAD = getattr(termios, "FIONREAD", None)
TIOCGWINSZ = getattr(termios, "TIOCGWINSZ", None)

# ------------ start of baudrate definitions ------------

# Add (possibly) missing baudrates (check termios man page) to termios


def add_baudrate_if_supported(dictionary: dict[int, int], rate: int) -> None:
    baudrate_name = "B%d" % rate
    if hasattr(termios, baudrate_name):
        dictionary[getattr(termios, baudrate_name)] = rate


# Check the termios man page (Line speed) to know where these
# values come from.
potential_baudrates = [
    0,
    110,
    115200,
    1200,
    134,
    150,
    1800,
    19200,
    200,
    230400,
    2400,
    300,
    38400,
    460800,
    4800,
    50,
    57600,
    600,
    75,
    9600,
]

ratedict: dict[int, int] = {}
for rate in potential_baudrates:
    add_baudrate_if_supported(ratedict, rate)

# Clean up variables to avoid unintended usage
del rate, add_baudrate_if_supported

# ------------ end of baudrate definitions ------------

delayprog = re.compile(b"\\$<([0-9]+)((?:/|\\*){0,2})>")

try:
    poll: type[select.poll] = select.poll
except AttributeError:
    # this is exactly the minimum necessary to support what we
    # do with poll objects
    class MinimalPoll:
        def __init__(self):
            pass

        def register(self, fd, flag):
            self.fd = fd

        # note: The 'timeout' argument is received as *milliseconds*
        def poll(self, timeout: float | None = None) -> list[int]:
            if timeout is None:
                r, w, e = select.select([self.fd], [], [])
            else:
                r, w, e = select.select([self.fd], [], [], timeout / 1000)
            return r

    poll = MinimalPoll  # type: ignore[assignment]


@dataclass(frozen=True, slots=True)
class UnixRefreshPlan:
    """Instructions for updating the terminal after a screen change.

    After the user types ``e`` to complete ``name``::

        Before: >>> def greet(nam|):
                                 ▲
                        LineUpdate here: insert_char "e"

         After: >>> def greet(name|):
                                  ▲

    Only the changed cells are sent to the terminal; unchanged rows
    are skipped entirely.
    """

    grow_lines: int
    """Number of blank lines to append at the bottom to accommodate new content."""
    use_tall_mode: bool
    """Use absolute cursor addressing via ``cup`` instead of relative moves.
    Activated when content exceeds one screen height."""
    offset: int
    """Vertical scroll offset: the buffer row displayed at the top of the terminal window."""
    reverse_scroll: int
    """Number of lines to scroll backwards (content moves down)."""
    forward_scroll: int
    """Number of lines to scroll forwards (content moves up)."""
    line_updates: tuple[LineUpdate, ...]
    cleared_lines: tuple[int, ...]
    """Row indices to erase (old content with no replacement)."""
    rendered_screen: RenderedScreen
    cursor: tuple[int, int]


class UnixConsole(Console):
    __buffer: list[_PendingWrite]
    __gone_tall: bool
    __move: _MoveFunc
    __offset: int

    def __init__(
        self,
        f_in: IO[bytes] | int = 0,
        f_out: IO[bytes] | int = 1,
        term: str = "",
        encoding: str = "",
    ):
        """
        Initialize the UnixConsole.

        Parameters:
        - f_in (int or file-like object): Input file descriptor or object.
        - f_out (int or file-like object): Output file descriptor or object.
        - term (str): Terminal name.
        - encoding (str): Encoding to use for I/O operations.
        """
        super().__init__(f_in, f_out, term, encoding)

        self.pollob = poll()
        self.pollob.register(self.input_fd, select.POLLIN)
        self.terminfo = terminfo.TermInfo(term or None)
        self.term = term
        self.is_apple_terminal = (
            platform.system() == "Darwin"
            and os.getenv("TERM_PROGRAM") == "Apple_Terminal"
        )

        try:
            self.__input_fd_set(tcgetattr(self.input_fd), ignore=frozenset())
        except _error as e:
            raise RuntimeError(f"termios failure ({e.args[1]})")

        @overload
        def _my_getstr(
            cap: str, optional: Literal[False] = False
        ) -> bytes: ...

        @overload
        def _my_getstr(cap: str, optional: bool) -> bytes | None: ...

        def _my_getstr(cap: str, optional: bool = False) -> bytes | None:
            r = self.terminfo.get(cap)
            if not optional and r is None:
                raise InvalidTerminal(
                    f"terminal doesn't have the required {cap} capability"
                )
            return r

        self._bel = _my_getstr("bel")
        self._civis = _my_getstr("civis", optional=True)
        self._clear = _my_getstr("clear")
        self._cnorm = _my_getstr("cnorm", optional=True)
        self._cub = _my_getstr("cub", optional=True)
        self._cub1 = _my_getstr("cub1", optional=True)
        self._cud = _my_getstr("cud", optional=True)
        self._cud1 = _my_getstr("cud1", optional=True)
        self._cuf = _my_getstr("cuf", optional=True)
        self._cuf1 = _my_getstr("cuf1", optional=True)
        self._cup = _my_getstr("cup")
        self._cuu = _my_getstr("cuu", optional=True)
        self._cuu1 = _my_getstr("cuu1", optional=True)
        self._dch1 = _my_getstr("dch1", optional=True)
        self._dch = _my_getstr("dch", optional=True)
        self._el = _my_getstr("el")
        self._hpa = _my_getstr("hpa", optional=True)
        self._ich = _my_getstr("ich", optional=True)
        self._ich1 = _my_getstr("ich1", optional=True)
        self._ind = _my_getstr("ind", optional=True)
        self._pad = _my_getstr("pad", optional=True)
        self._ri = _my_getstr("ri", optional=True)
        self._rmkx = _my_getstr("rmkx", optional=True)
        self._smkx = _my_getstr("smkx", optional=True)

        self.__setup_movement()

        self.event_queue = EventQueue(
            self.input_fd, self.encoding, self.terminfo
        )
        self.cursor_visible = True

        signal.signal(signal.SIGCONT, self._sigcont_handler)

    def _sigcont_handler(self, signum, frame):
        self.restore()
        self.prepare()

    def __read(self, n: int) -> bytes:
        return os.read(self.input_fd, n)

    def change_encoding(self, encoding: str) -> None:
        """
        Change the encoding used for I/O operations.

        Parameters:
        - encoding (str): New encoding to use.
        """
        self.encoding = encoding

    def refresh(self, rendered_screen: RenderedScreen) -> None:
        """
        Refresh the console screen.

        Parameters:
        - rendered_screen: Structured rendered screen contents and cursor.
        """
        c_xy = rendered_screen.cursor
        trace(
            "unix.refresh start cursor={cursor} lines={lines} prev_lines={prev_lines} "
            "offset={offset} posxy={posxy}",
            cursor=c_xy,
            lines=len(rendered_screen.composed_lines),
            prev_lines=len(self._rendered_screen.composed_lines),
            offset=self.__offset,
            posxy=self.posxy,
        )
        plan = self.__plan_refresh(rendered_screen, c_xy)
        self.__apply_refresh_plan(plan)

    def __plan_refresh(
        self,
        rendered_screen: RenderedScreen,
        c_xy: tuple[int, int],
    ) -> UnixRefreshPlan:
        cx, cy = c_xy
        height = self.height
        old_offset = offset = self.__offset
        prev_composed = self._rendered_screen.composed_lines
        previous_lines = list(prev_composed)
        next_lines = list(rendered_screen.composed_lines)
        line_count = len(next_lines)

        grow_lines = 0
        if not self.__gone_tall:
            grow_lines = max(
                min(line_count, height) - len(prev_composed),
                0,
            )
            previous_lines.extend([EMPTY_RENDER_LINE] * grow_lines)
        elif len(previous_lines) < line_count:
            previous_lines.extend([EMPTY_RENDER_LINE] * (line_count - len(previous_lines)))

        use_tall_mode = self.__gone_tall or line_count > height

        # we make sure the cursor is on the screen, and that we're
        # using all of the screen if we can
        if cy < offset:
            offset = cy
        elif cy >= offset + height:
            offset = cy - height + 1
        elif offset > 0 and line_count < offset + height:
            offset = max(line_count - height, 0)
            next_lines.append(EMPTY_RENDER_LINE)

        oldscr = previous_lines[old_offset : old_offset + height]
        newscr = next_lines[offset : offset + height]

        reverse_scroll = 0
        forward_scroll = 0
        if old_offset > offset and self._ri:
            reverse_scroll = old_offset - offset
            for _ in range(reverse_scroll):
                if oldscr:
                    oldscr.pop(-1)
                oldscr.insert(0, EMPTY_RENDER_LINE)
        elif old_offset < offset and self._ind:
            forward_scroll = offset - old_offset
            for _ in range(forward_scroll):
                if oldscr:
                    oldscr.pop(0)
                oldscr.append(EMPTY_RENDER_LINE)

        line_updates: list[LineUpdate] = []
        px, _ = self.posxy
        for y, oldline, newline in zip(range(offset, offset + height), oldscr, newscr):
            update = self.__plan_changed_line(y, oldline, newline, px)
            if update is not None:
                line_updates.append(update)

        cleared_lines = tuple(range(offset + len(newscr), offset + len(oldscr)))
        console_rendered_screen = RenderedScreen(tuple(next_lines), c_xy)
        trace(
            "unix.refresh plan grow={grow} tall={tall} offset={offset} "
            "reverse_scroll={reverse_scroll} forward_scroll={forward_scroll} "
            "updates={updates} clears={clears}",
            grow=grow_lines,
            tall=use_tall_mode,
            offset=offset,
            reverse_scroll=reverse_scroll,
            forward_scroll=forward_scroll,
            updates=len(line_updates),
            clears=len(cleared_lines),
        )
        return UnixRefreshPlan(
            grow_lines=grow_lines,
            use_tall_mode=use_tall_mode,
            offset=offset,
            reverse_scroll=reverse_scroll,
            forward_scroll=forward_scroll,
            line_updates=tuple(line_updates),
            cleared_lines=cleared_lines,
            rendered_screen=console_rendered_screen,
            cursor=(cx, cy),
        )

    def __apply_refresh_plan(self, plan: UnixRefreshPlan) -> None:
        cx, cy = plan.cursor
        trace(
            "unix.refresh apply cursor={cursor} updates={updates} clears={clears}",
            cursor=plan.cursor,
            updates=len(plan.line_updates),
            clears=len(plan.cleared_lines),
        )
        visual_style = self.begin_redraw_visualization()
        screen_line_count = len(self._rendered_screen.composed_lines)

        for _ in range(plan.grow_lines):
            self.__hide_cursor()
            if screen_line_count:
                self.__move(0, screen_line_count - 1)
                self.__write("\n")
            self.posxy = 0, screen_line_count
            screen_line_count += 1

        if plan.use_tall_mode and not self.__gone_tall:
            self.__gone_tall = True
            self.__move = self.__move_tall

        old_offset = self.__offset
        if plan.reverse_scroll:
            self.__hide_cursor()
            self.__write_code(self._cup, 0, 0)
            self.posxy = 0, old_offset
            for _ in range(plan.reverse_scroll):
                self.__write_code(self._ri)
        elif plan.forward_scroll:
            self.__hide_cursor()
            self.__write_code(self._cup, self.height - 1, 0)
            self.posxy = 0, old_offset + self.height - 1
            for _ in range(plan.forward_scroll):
                self.__write_code(self._ind)

        self.__offset = plan.offset

        for update in plan.line_updates:
            self.__apply_line_update(update, visual_style)

        for y in plan.cleared_lines:
            self.__hide_cursor()
            self.__move(0, y)
            self.posxy = 0, y
            self.__write_code(self._el)

        self.__show_cursor()
        self.move_cursor(cx, cy)
        self.flushoutput()
        self.sync_rendered_screen(plan.rendered_screen, self.posxy)

    def move_cursor(self, x: int, y: int) -> None:
        """
        Move the cursor to the specified position on the screen.

        Parameters:
        - x (int): X coordinate.
        - y (int): Y coordinate.
        """
        if y < self.__offset or y >= self.__offset + self.height:
            trace(
                "unix.move_cursor offscreen x={x} y={y} offset={offset} height={height}",
                x=x,
                y=y,
                offset=self.__offset,
                height=self.height,
            )
            self.event_queue.insert(Event("scroll", ""))
        else:
            trace("unix.move_cursor x={x} y={y}", x=x, y=y)
            self.__move(x, y)
            self.posxy = x, y
            self.flushoutput()

    def prepare(self) -> None:
        """
        Prepare the console for input/output operations.
        """
        trace("unix.prepare")
        self.__buffer = []

        self.__svtermstate = tcgetattr(self.input_fd)
        raw = self.__svtermstate.copy()
        raw.iflag &= ~(termios.INPCK | termios.ISTRIP | termios.IXON)
        raw.oflag &= ~(termios.OPOST)
        raw.cflag &= ~(termios.CSIZE | termios.PARENB)
        raw.cflag |= termios.CS8
        raw.iflag |= termios.BRKINT
        raw.lflag &= ~(termios.ICANON | termios.ECHO | termios.IEXTEN)
        raw.lflag |= termios.ISIG
        raw.cc[termios.VMIN] = b"\x01"
        raw.cc[termios.VTIME] = b"\x00"
        self.__input_fd_set(raw)
        self.__rawtermstate = raw

        # Apple Terminal will re-wrap lines for us unless we preempt the
        # damage.
        if self.is_apple_terminal:
            os.write(self.output_fd, b"\033[?7l")

        self.height, self.width = self.getheightwidth()

        self.posxy = 0, 0
        self.__gone_tall = False
        self.__move = self.__move_short
        self.__offset = 0
        self.sync_rendered_screen(RenderedScreen.empty(), self.posxy)

        self.__maybe_write_code(self._smkx)

        try:
            self.old_sigwinch = signal.signal(signal.SIGWINCH, self.__sigwinch)
        except ValueError:
            pass

        self.__enable_bracketed_paste()

    def restore(self) -> None:
        """
        Restore the console to the default state
        """
        trace("unix.restore")
        self.__write(ANSIColors.RESET)
        self.__disable_bracketed_paste()
        self.__maybe_write_code(self._rmkx)
        self.flushoutput()
        self.__input_fd_set(self.__svtermstate)

        if self.is_apple_terminal:
            os.write(self.output_fd, b"\033[?7h")

        if hasattr(self, "old_sigwinch"):
            try:
                signal.signal(signal.SIGWINCH, self.old_sigwinch)
            except ValueError as e:
                import threading
                if threading.current_thread() is threading.main_thread():
                    raise e
            del self.old_sigwinch

    def push_char(self, char: int | bytes) -> None:
        """
        Push a character to the console event queue.
        """
        trace("push char {char!r}", char=char)
        self.event_queue.push(char)

    def get_event(self, block: bool = True) -> Event | None:
        """
        Get an event from the console event queue.

        Parameters:
        - block (bool): Whether to block until an event is available.

        Returns:
        - Event: Event object from the event queue.
        """
        if not block and not self.wait(timeout=0):
            return None

        while self.event_queue.empty():
            while True:
                try:
                    self.push_char(self.__read(1))
                except OSError as err:
                    if err.errno == errno.EINTR:
                        if not self.event_queue.empty():
                            return self.event_queue.get()
                        else:
                            continue
                    elif err.errno == errno.EIO:
                        raise SystemExit(errno.EIO)
                    else:
                        raise
                else:
                    break
        return self.event_queue.get()

    def wait(self, timeout: float | None = None) -> bool:
        """
        Wait for events on the console.
        """
        return (
            not self.event_queue.empty()
            or bool(self.pollob.poll(timeout))
        )

    def set_cursor_vis(self, visible: bool) -> None:
        """
        Set the visibility of the cursor.

        Parameters:
        - visible (bool): Visibility flag.
        """
        if visible:
            self.__show_cursor()
        else:
            self.__hide_cursor()

    if TIOCGWINSZ:

        def getheightwidth(self):
            """
            Get the height and width of the console.

            Returns:
            - tuple: Height and width of the console.
            """
            try:
                return int(os.environ["LINES"]), int(os.environ["COLUMNS"])
            except (KeyError, TypeError, ValueError):
                try:
                    size = ioctl(self.input_fd, TIOCGWINSZ, b"\000" * 8)
                except OSError:
                    return 25, 80
                height, width = struct.unpack("hhhh", size)[0:2]
                if not height:
                    return 25, 80
                return height, width

    else:

        def getheightwidth(self):
            """
            Get the height and width of the console.

            Returns:
            - tuple: Height and width of the console.
            """
            try:
                return int(os.environ["LINES"]), int(os.environ["COLUMNS"])
            except (KeyError, TypeError, ValueError):
                return 25, 80

    def forgetinput(self):
        """
        Discard any pending input on the console.
        """
        termios.tcflush(self.input_fd, termios.TCIFLUSH)

    def flushoutput(self):
        """
        Flush the output buffer.
        """
        for text, iscode in self.__buffer:
            if iscode:
                self.__tputs(text)
            else:
                os.write(self.output_fd, text.encode(self.encoding, "replace"))
        del self.__buffer[:]

    def finish(self):
        """
        Finish console operations and flush the output buffer.
        """
        rendered_lines = self._rendered_screen.composed_lines
        y = len(rendered_lines) - 1
        while y >= 0 and not rendered_lines[y].text:
            y -= 1
        self.__move(0, min(y, self.height + self.__offset - 1))
        self.__write(ANSIColors.RESET)
        self.__write("\n\r")
        self.flushoutput()

    def beep(self):
        """
        Emit a beep sound.
        """
        self.__maybe_write_code(self._bel)
        self.flushoutput()

    if FIONREAD:

        def getpending(self):
            """
            Get pending events from the console event queue.

            Returns:
            - Event: Pending event from the event queue.
            """
            e = Event("key", "", b"")

            while not self.event_queue.empty():
                e2 = self.event_queue.get()
                e.data += e2.data
                e.raw += e2.raw

            amount = struct.unpack("i", ioctl(self.input_fd, FIONREAD, b"\0\0\0\0"))[0]
            trace("getpending({a})", a=amount)
            raw = self.__read(amount)
            data = str(raw, self.encoding, "replace")
            e.data += data
            e.raw += raw
            return e

    else:

        def getpending(self):
            """
            Get pending events from the console event queue.

            Returns:
            - Event: Pending event from the event queue.
            """
            e = Event("key", "", b"")

            while not self.event_queue.empty():
                e2 = self.event_queue.get()
                e.data += e2.data
                e.raw += e2.raw

            amount = 10000
            raw = self.__read(amount)
            data = str(raw, self.encoding, "replace")
            e.data += data
            e.raw += raw
            return e

    def clear(self):
        """
        Clear the console screen.
        """
        trace("unix.clear")
        self.__write_code(self._clear)
        self.__gone_tall = True
        self.__move = self.__move_tall
        self.posxy = 0, 0
        self.sync_rendered_screen(RenderedScreen.empty(), self.posxy)

    @property
    def input_hook(self):
        # avoid inline imports here so the repl doesn't get flooded
        # with import logging from -X importtime=2
        if posix is not None and posix._is_inputhook_installed():
            return self.__run_input_hook

    def __run_input_hook(self):
        # gh-152907: input hooks expect cooked output, but pyrepl runs with
        # OPOST disabled.  Restore the saved output flags around the hook
        # (only oflag; input must stay raw at the prompt).
        cooked = self.__rawtermstate.copy()
        cooked.oflag = self.__svtermstate.oflag
        self.__input_fd_set(cooked)
        try:
            return posix._inputhook()
        finally:
            self.__input_fd_set(self.__rawtermstate)

    def __enable_bracketed_paste(self) -> None:
        os.write(self.output_fd, b"\x1b[?2004h")

    def __disable_bracketed_paste(self) -> None:
        os.write(self.output_fd, b"\x1b[?2004l")

    def __setup_movement(self):
        """
        Set up the movement functions based on the terminal capabilities.
        """
        if 0 and self._hpa:  # hpa don't work in windows telnet :-(
            self.__move_x = self.__move_x_hpa
        elif self._cub and self._cuf:
            self.__move_x = self.__move_x_cub_cuf
        elif self._cub1 and self._cuf1:
            self.__move_x = self.__move_x_cub1_cuf1
        else:
            raise RuntimeError("insufficient terminal (horizontal)")

        if self._cuu and self._cud:
            self.__move_y = self.__move_y_cuu_cud
        elif self._cuu1 and self._cud1:
            self.__move_y = self.__move_y_cuu1_cud1
        else:
            raise RuntimeError("insufficient terminal (vertical)")

        if self._dch1:
            self.dch1 = self._dch1
        elif self._dch:
            self.dch1 = terminfo.tparm(self._dch, 1)
        else:
            self.dch1 = None

        if self._ich1:
            self.ich1 = self._ich1
        elif self._ich:
            self.ich1 = terminfo.tparm(self._ich, 1)
        else:
            self.ich1 = None

        self.__move = self.__move_short

    @staticmethod
    def __cell_index_from_x(line: RenderLine, x_coord: int) -> int:
        width = 0
        index = 0
        while index < len(line.cells) and width < x_coord:
            width += line.cells[index].width
            index += 1
        return index

    def __plan_changed_line(
        self,
        y: int,
        oldline: RenderLine,
        newline: RenderLine,
        px_coord: int,
    ) -> LineUpdate | None:
        # NOTE: The shared replace_char / replace_span / rewrite_suffix logic
        # is duplicated in WindowsConsole.__plan_changed_line. Keep changes to
        # these common cases synchronised between the two files. Yes, this is
        # duplicated on purpose; the two backends agree just enough to make a
        # shared helper a trap. Unix-only cases (insert_char, delete_then_insert)
        # rely on terminal capabilities (ich1/dch1) that are unavailable on
        # Windows.
        diff = diff_render_lines(oldline, newline)
        if diff is None:
            return None

        start_cell = diff.start_cell
        start_x = diff.start_x

        if (
            self.ich1
            and not diff.old_cells
            and (visible_new_cells := tuple(
                cell for cell in diff.new_cells if cell.width
            ))
            and len(visible_new_cells) == 1
            and all(cell.width == 0 for cell in diff.new_cells[1:])
            and oldline.cells[start_cell:] == newline.cells[start_cell + 1 :]
        ):
            px_cell = self.__cell_index_from_x(oldline, px_coord)
            if (
                y == self.posxy[1]
                and start_x > self.posxy[0]
                and oldline.cells[px_cell:start_cell]
                == newline.cells[px_cell + 1 : start_cell + 1]
            ):
                start_cell = px_cell
                start_x = px_coord
            planned_cells = diff.new_cells
            changed_cell = visible_new_cells[0]
            return LineUpdate(
                kind="insert_char",
                y=y,
                start_cell=start_cell,
                start_x=start_x,
                cells=planned_cells,
                char_width=changed_cell.width,
                reset_to_margin=requires_cursor_resync(planned_cells),
            )

        if (
            len(diff.old_cells) == 1
            and len(diff.new_cells) == 1
            and diff.old_cells[0].width == diff.new_cells[0].width
        ):
            planned_cells = diff.new_cells
            changed_cell = planned_cells[0]
            return LineUpdate(
                kind="replace_char",
                y=y,
                start_cell=start_cell,
                start_x=start_x,
                cells=planned_cells,
                char_width=changed_cell.width,
                reset_to_margin=requires_cursor_resync(planned_cells),
            )

        if diff.old_changed_width == diff.new_changed_width:
            planned_cells = diff.new_cells
            return LineUpdate(
                kind="replace_span",
                y=y,
                start_cell=start_cell,
                start_x=start_x,
                cells=planned_cells,
                char_width=diff.new_changed_width,
                reset_to_margin=requires_cursor_resync(planned_cells),
            )

        if (
            self.dch1
            and self.ich1
            and newline.width == self.width
            and start_x < newline.width - 2
            and newline.cells[start_cell + 1 : -1] == oldline.cells[start_cell:-2]
        ):
            planned_cells = (newline.cells[start_cell],)
            changed_cell = planned_cells[0]
            return LineUpdate(
                kind="delete_then_insert",
                y=y,
                start_cell=start_cell,
                start_x=start_x,
                cells=planned_cells,
                char_width=changed_cell.width,
                reset_to_margin=requires_cursor_resync(planned_cells),
            )

        suffix_cells = newline.cells[start_cell:]
        return LineUpdate(
            kind="rewrite_suffix",
            y=y,
            start_cell=start_cell,
            start_x=start_x,
            cells=suffix_cells,
            char_width=sum(cell.width for cell in suffix_cells),
            clear_eol=oldline.width > newline.width,
            reset_to_margin=requires_cursor_resync(suffix_cells),
        )

    def __apply_line_update(
        self,
        update: LineUpdate,
        visual_style: str | None = None,
    ) -> None:
        text = render_cells(update.cells, visual_style) if visual_style else update.text
        trace(
            "unix.refresh update kind={kind} y={y} x={x} text={text} "
            "clear_eol={clear_eol} reset_to_margin={reset}",
            kind=update.kind,
            y=update.y,
            x=update.start_x,
            text=trace_text(text),
            clear_eol=update.clear_eol,
            reset=update.reset_to_margin,
        )
        if update.kind == "insert_char":
            self.__move(update.start_x, update.y)
            self.__write_code(self.ich1)
            self.__write(text)
            self.posxy = update.start_x + update.char_width, update.y
        elif update.kind in {"replace_char", "replace_span"}:
            self.__move(update.start_x, update.y)
            self.__write(text)
            self.posxy = update.start_x + update.char_width, update.y
        elif update.kind == "delete_then_insert":
            self.__hide_cursor()
            self.__move(self.width - 2, update.y)
            self.posxy = self.width - 2, update.y
            self.__write_code(self.dch1)
            self.__move(update.start_x, update.y)
            self.__write_code(self.ich1)
            self.__write(text)
            self.posxy = update.start_x + update.char_width, update.y
        else:
            self.__hide_cursor()
            self.__move(update.start_x, update.y)
            if update.clear_eol:
                self.__write_code(self._el)
            self.__write(text)
            self.posxy = update.start_x + update.char_width, update.y

        if update.reset_to_margin:
            # Non-SGR terminal controls can affect the cursor position.
            self.move_cursor(0, update.y)

    def __write(self, text):
        self.__buffer.append((text, False))

    def __write_code(self, fmt, *args):
        self.__buffer.append((terminfo.tparm(fmt, *args), True))

    def __maybe_write_code(self, fmt, *args):
        if fmt:
            self.__write_code(fmt, *args)

    def __move_y_cuu1_cud1(self, y):
        assert self._cud1 is not None
        assert self._cuu1 is not None
        dy = y - self.posxy[1]
        if dy > 0:
            self.__write_code(dy * self._cud1)
        elif dy < 0:
            self.__write_code((-dy) * self._cuu1)

    def __move_y_cuu_cud(self, y):
        dy = y - self.posxy[1]
        if dy > 0:
            self.__write_code(self._cud, dy)
        elif dy < 0:
            self.__write_code(self._cuu, -dy)

    def __move_x_hpa(self, x: int) -> None:
        if x != self.posxy[0]:
            self.__write_code(self._hpa, x)

    def __move_x_cub1_cuf1(self, x: int) -> None:
        assert self._cuf1 is not None
        assert self._cub1 is not None
        dx = x - self.posxy[0]
        if dx > 0:
            self.__write_code(self._cuf1 * dx)
        elif dx < 0:
            self.__write_code(self._cub1 * (-dx))

    def __move_x_cub_cuf(self, x: int) -> None:
        dx = x - self.posxy[0]
        if dx > 0:
            self.__write_code(self._cuf, dx)
        elif dx < 0:
            self.__write_code(self._cub, -dx)

    def __move_short(self, x, y):
        self.__move_x(x)
        self.__move_y(y)

    def __move_tall(self, x, y):
        assert 0 <= y - self.__offset < self.height, y - self.__offset
        self.__write_code(self._cup, y - self.__offset, x)

    def __sigwinch(self, signum, frame):
        self.event_queue.insert(Event("resize", ""))

    def __hide_cursor(self):
        if self.cursor_visible:
            self.__maybe_write_code(self._civis)
            self.cursor_visible = False

    def __show_cursor(self):
        if not self.cursor_visible:
            self.__maybe_write_code(self._cnorm)
            self.cursor_visible = True

    def repaint(self):
        composed = self._rendered_screen.composed_lines
        trace(
            "unix.repaint gone_tall={gone_tall} screen_lines={lines} offset={offset}",
            gone_tall=self.__gone_tall,
            lines=len(composed),
            offset=self.__offset,
        )
        if not self.__gone_tall:
            self.posxy = 0, self.posxy[1]
            self.__write("\r")
            ns = len(composed) * ["\000" * self.width]
        else:
            self.posxy = 0, self.__offset
            self.__move(0, self.__offset)
            ns = self.height * ["\000" * self.width]
        self.sync_rendered_screen(
            RenderedScreen.from_screen_lines(ns, self.posxy),
            self.posxy,
        )

    def __tputs(self, fmt, prog=delayprog):
        """A Python implementation of the curses tputs function; the
        curses one can't really be wrapped in a sane manner.

        I have the strong suspicion that this is complexity that
        will never do anyone any good."""
        # using .get() means that things will blow up
        # only if the bps is actually needed (which I'm
        # betting is pretty unlikely)
        bps = ratedict.get(self.__svtermstate.ospeed)
        while True:
            m = prog.search(fmt)
            if not m:
                os.write(self.output_fd, fmt)
                break
            x, y = m.span()
            os.write(self.output_fd, fmt[:x])
            fmt = fmt[y:]
            delay = int(m.group(1))
            if b"*" in m.group(2):
                delay *= self.height
            if self._pad and bps is not None:
                nchars = (bps * delay) / 1000
                os.write(self.output_fd, self._pad * nchars)
            else:
                time.sleep(float(delay) / 1000.0)

    def __input_fd_set(
        self,
        state: TermState,
        ignore: AbstractSet[int] = _error_codes_to_ignore,
    ) -> bool:
        try:
            tcsetattr(self.input_fd, termios.TCSADRAIN, state)
        except termios.error as te:
            if te.args[0] not in ignore:
                raise
            return False
        else:
            return True