path: root/tests/extern/kitty/screen.py

"""ctypes wrapper providing the Kitty ``Screen`` API on top of our C terminal.

The shared library ``libterm.so`` is loaded via :mod:`ctypes`. Python calls
only the ``kitty_*`` functions of ``screen_bridge.c``, which wrap the public
API (declared in ``src/term/term.h``). Python classes on top of them match the
subset of Kitty's test API used by ``parser.py``.
"""

import ctypes
import os

# ---------------------------------------------------------------------------
# Shared library loading
# ---------------------------------------------------------------------------

_lib = ctypes.CDLL(os.environ.get('WSAP_KITTY_LIB', os.path.join(os.path.dirname(__file__), 'libterm.so')))

# struct term_cell { uint32_t text, fg, bg; uint16_t attrs; uint8_t width, reserved; }
class TermCell(ctypes.Structure):
    _fields_ = [
        ('text', ctypes.c_uint32),
        ('fg', ctypes.c_uint32),
        ('bg', ctypes.c_uint32),
        ('attrs', ctypes.c_uint16),
        ('width', ctypes.c_uint8),
        ('reserved', ctypes.c_uint8),
    ]


assert ctypes.sizeof(TermCell) == 16

# Reply callback type: void (*)(void *ud, const char *bytes, size_t len).
# The bytes are not NUL-terminated, so the callback takes a raw pointer.
REPLY_FN = ctypes.CFUNCTYPE(None, ctypes.c_void_p, ctypes.POINTER(ctypes.c_char), ctypes.c_size_t)

# Function signatures
_lib.kitty_term_new.restype = ctypes.c_void_p
_lib.kitty_term_new.argtypes = [ctypes.c_int, ctypes.c_int, REPLY_FN, ctypes.c_void_p]
_lib.kitty_term_free.restype = None
_lib.kitty_term_free.argtypes = [ctypes.c_void_p]
_lib.kitty_term_resize.restype = ctypes.c_bool
_lib.kitty_term_resize.argtypes = [ctypes.c_void_p, ctypes.c_int, ctypes.c_int]
_lib.kitty_term_write.restype = None
_lib.kitty_term_write.argtypes = [ctypes.c_void_p, ctypes.c_char_p, ctypes.c_size_t]
_lib.kitty_term_cell.restype = ctypes.c_bool
_lib.kitty_term_cell.argtypes = [ctypes.c_void_p, ctypes.c_int, ctypes.c_int,
                                 ctypes.POINTER(TermCell)]
_lib.kitty_term_cell_text.restype = ctypes.c_int
_lib.kitty_term_cell_text.argtypes = [ctypes.c_void_p, ctypes.c_int, ctypes.c_int,
                                      ctypes.POINTER(ctypes.c_uint32), ctypes.c_int]
for _name, _type in {
    'cols': ctypes.c_int,
    'rows': ctypes.c_int,
    'cursor_x': ctypes.c_int,
    'cursor_y': ctypes.c_int,
    'cursor_style': ctypes.c_int,
    'cursor_blink': ctypes.c_bool,
    'application_cursor': ctypes.c_bool,
}.items():
    _fn = getattr(_lib, 'kitty_term_' + _name)
    _fn.argtypes = [ctypes.c_void_p]
    _fn.restype = _type

# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------

TERM_DEFAULT_FG = 0xFFFFFFFF
TERM_DEFAULT_BG = 0xFF1E1E2E

# Cell attributes (must match enum term_attr in term.h)
TERM_ATTR_BOLD            = 1 << 0
TERM_ATTR_ITALIC          = 1 << 1
TERM_ATTR_FAINT           = 1 << 2
TERM_ATTR_STRIKE          = 1 << 3
TERM_ATTR_OVERLINE        = 1 << 4
TERM_ATTR_LINK            = 1 << 5
TERM_ATTR_EMOJI           = 1 << 6
TERM_ATTR_TEXT            = 1 << 7
TERM_ATTR_UNDERLINE_SHIFT = 8
TERM_ATTR_UNDERLINE_MASK  = 7 << TERM_ATTR_UNDERLINE_SHIFT

# The longest grapheme cluster that a cell's text is read with.
_CLUSTER_CAP = 32

# Test-only accessors compiled against the current C terminal definition.
for _name, _type in {
    'scroll_top': ctypes.c_int,
    'scroll_bottom': ctypes.c_int,
    'cur_fg': ctypes.c_uint32,
    'cur_bg': ctypes.c_uint32,
    'bold': ctypes.c_bool,
    'faint': ctypes.c_bool,
    'italic': ctypes.c_bool,
    'strike': ctypes.c_bool,
    'inverse': ctypes.c_bool,
    'underline': ctypes.c_uint8,
}.items():
    _fn = getattr(_lib, 'kitty_screen_' + _name)
    _fn.argtypes = [ctypes.c_void_p]
    _fn.restype = _type

_lib.kitty_screen_row_length.argtypes = [ctypes.c_void_p, ctypes.c_int]
_lib.kitty_screen_row_length.restype = ctypes.c_int
_lib.kitty_screen_cell_is_tab.argtypes = [ctypes.c_void_p, ctypes.c_int, ctypes.c_int]
_lib.kitty_screen_cell_is_tab.restype = ctypes.c_bool

# ANSI color table from escape.c — used for reverse-mapping to Kitty indices.
# Must match the built-in default palette in src/term/palette.c.
_ANSI_COLORS = [
    0xFF585B70, 0xFFCC6666, 0xFFB5BD68, 0xFFF0C674,
    0xFF81A2BE, 0xFFB294BB, 0xFF8ABEB7, 0xFFC5C8C6,
    0xFF666666, 0xFFD54E53, 0xFFB9CA4A, 0xFFE7C547,
    0xFF7AA6DA, 0xFFC397D8, 0xFF70C0B1, 0xFFEAEAEA,
]


def _build_color_table():
    """Build a reverse lookup from RGB (0xFFRRGGBB) to Kitty indexed color."""
    table = {}
    for i, rgb in enumerate(_ANSI_COLORS):
        table[rgb] = (i << 8) | 1
    levels = [0, 95, 135, 175, 215, 255]
    for i in range(16, 232):
        n = i - 16
        r, g, b = levels[n // 36], levels[(n // 6) % 6], levels[n % 6]
        rgb = 0xFF000000 | (r << 16) | (g << 8) | b
        table[rgb] = (i << 8) | 1
    for i in range(232, 256):
        gray = 8 + (i - 232) * 10
        rgb = 0xFF000000 | (gray << 16) | (gray << 8) | gray
        table[rgb] = (i << 8) | 1
    return table


_COLOR_TO_KITTY = _build_color_table()


TERM_COLOR_REF = 0x01000000


def _convert_color(color):
    """Convert our cell color to Kitty's format.

    Ours: a palette reference (top byte 0x01; slot 0-255, 0x100 default fg,
    0x101 default bg) or a literal 0xFFRRGGBB color.

    Kitty format:
      0           = default
      index<<8|1  = indexed (0-255)
      r<<24|g<<16|b<<8|2 = RGB
    """
    if (color >> 24) == (TERM_COLOR_REF >> 24):
        slot = color & 0x1FF
        return 0 if slot >= 0x100 else (slot << 8) | 1
    if color == TERM_DEFAULT_FG or color == TERM_DEFAULT_BG:
        return 0
    kitty = _COLOR_TO_KITTY.get(color)
    if kitty is not None:
        return kitty
    r = (color >> 16) & 0xFF
    g = (color >> 8) & 0xFF
    b = color & 0xFF
    return (r << 24) | (g << 16) | (b << 8) | 2


# ---------------------------------------------------------------------------
# Screen property readers
# ---------------------------------------------------------------------------

def _read_property(ptr, name):
    return getattr(_lib, 'kitty_screen_' + name)(ptr)


# ---------------------------------------------------------------------------
# Cursor — wraps the terminal's current cursor SGR state + position
# ---------------------------------------------------------------------------

class Cursor:
    """Wraps the terminal's current cursor state (SGR attributes + position)."""

    def __init__(self, screen):
        self._screen = screen

    @property
    def _ptr(self):
        return self._screen._ptr

    # -- position --
    @property
    def x(self):
        return _lib.kitty_term_cursor_x(self._screen._term)

    @property
    def y(self):
        return _lib.kitty_term_cursor_y(self._screen._term)

    # -- SGR boolean attributes --
    @property
    def bold(self):
        return _read_property(self._ptr, 'bold')

    @property
    def italic(self):
        return _read_property(self._ptr, 'italic')

    @property
    def dim(self):
        return _read_property(self._ptr, 'faint')

    @property
    def reverse(self):
        return _read_property(self._ptr, 'inverse')

    @property
    def strikethrough(self):
        return _read_property(self._ptr, 'strike')

    @property
    def decoration(self):
        # Kitty: 0=none, 1=straight, 2=double, 3=curly, 4=dotted, 5=dashed
        return _read_property(self._ptr, 'underline')

    # -- colors --
    @property
    def fg(self):
        return _convert_color(_read_property(self._ptr, 'cur_fg'))

    @property
    def bg(self):
        return _convert_color(_read_property(self._ptr, 'cur_bg'))

    # -- cursor style --
    @property
    def blink(self):
        return _lib.kitty_term_cursor_blink(self._screen._term)

    @property
    def shape(self):
        return _lib.kitty_term_cursor_style(self._screen._term)


# ---------------------------------------------------------------------------
# CellCursor — wraps a single cell's attributes
# ---------------------------------------------------------------------------

class CellCursor:
    """Wraps a single cell's attributes for SGR inspection."""

    def __init__(self, cell):
        self._cell = cell

    @property
    def bold(self):
        return bool(self._cell.attrs & TERM_ATTR_BOLD)

    @property
    def italic(self):
        return bool(self._cell.attrs & TERM_ATTR_ITALIC)

    @property
    def dim(self):
        return bool(self._cell.attrs & TERM_ATTR_FAINT)

    @property
    def reverse(self):
        # Our terminal swaps fg/bg when inverse is set, so cells don't
        # carry a separate inverse flag.
        return False

    @property
    def strikethrough(self):
        return bool(self._cell.attrs & TERM_ATTR_STRIKE)

    @property
    def decoration(self):
        # enum term_underline uses Kitty's numbers: 0=none, 1=straight,
        # 2=double, 3=curly, 4=dotted, 5=dashed.
        return (self._cell.attrs & TERM_ATTR_UNDERLINE_MASK) >> TERM_ATTR_UNDERLINE_SHIFT

    @property
    def fg(self):
        return _convert_color(self._cell.fg)

    @property
    def bg(self):
        return _convert_color(self._cell.bg)


# ---------------------------------------------------------------------------
# Line — wraps a screen line, providing cell access
# ---------------------------------------------------------------------------

class Line:
    """Wraps a screen line, providing cell access and text content."""

    def __init__(self, screen, y):
        self._screen = screen
        self._y = y

    def cursor_from(self, x):
        """Return a :class:`CellCursor` for cell (x, y)."""
        cell = TermCell()
        if not _lib.kitty_term_cell(self._screen._term, x, self._y, ctypes.byref(cell)):
            return CellCursor(TermCell())
        return CellCursor(cell)

    def __str__(self):
        """Return the text content of the line up to the written length."""
        length = min(_lib.kitty_screen_row_length(self._screen._term, self._y),
                     self._screen.columns)
        if length == 0:
            return ''
        result = []
        cell = TermCell()
        cps = (ctypes.c_uint32 * _CLUSTER_CAP)()
        for x in range(length):
            if not _lib.kitty_term_cell(self._screen._term, x, self._y, ctypes.byref(cell)):
                result.append(' ')
                continue
            # Width 0 is the right half of a wide glyph.
            if cell.width == 0:
                continue
            if _lib.kitty_screen_cell_is_tab(self._screen._term, x, self._y):
                result.append('\t')
                continue
            n = _lib.kitty_term_cell_text(self._screen._term, x, self._y, cps, _CLUSTER_CAP)
            n = min(n, _CLUSTER_CAP)
            result.append(''.join(chr(cp) for cp in cps[:n]))
        return ''.join(result)


# ---------------------------------------------------------------------------
# Callbacks — mock callbacks object for OSC/DCS responses
# ---------------------------------------------------------------------------

class Callbacks:
    """Mock callbacks object for OSC/DCS responses."""

    def __init__(self, screen=None):
        self._screen = screen
        self.wtcbuf = b''
        self.titlebuf = []
        self.iconbuf = ''
        self.colorbuf = ''
        self.notifications = []
        self.printbuf = []

    def clear(self):
        self.wtcbuf = b''
        self.titlebuf = []
        self.iconbuf = ''
        self.colorbuf = ''
        self.notifications = []
        self.printbuf = []
        if self._screen is not None:
            self._screen._response_buf.clear()


# ---------------------------------------------------------------------------
# Screen — main screen wrapper using ctypes
# ---------------------------------------------------------------------------

class Screen:
    """Main screen wrapper using ctypes to drive the C terminal."""

    def __init__(self, cols=80, rows=5):
        self._cols = cols
        self._rows = rows
        self._response_buf = bytearray()
        self._write_buf = bytearray()
        self._callbacks = Callbacks(self)
        # Keep a reference to the callback function so it isn't GC'd
        self._resp_fn = REPLY_FN(self._on_response)
        self._term = _lib.kitty_term_new(cols, rows, self._resp_fn, None)
        if not self._term:
            raise RuntimeError('kitty_term_new failed')

    @property
    def _ptr(self):
        return ctypes.cast(self._term, ctypes.c_void_p).value

    # -- response callback --
    def _on_response(self, ud, data, length):
        if data and length:
            self._response_buf.extend(data[:length])
        self._callbacks.wtcbuf = bytes(self._response_buf)

    # -- writing --
    def write(self, data):
        if data:
            _lib.kitty_term_write(self._term, bytes(data), len(data))

    # -- test write buffer API (matches Kitty's test_parse interface) --
    def test_create_write_buffer(self):
        """Return the available space in the write buffer (int, falsy when full)."""
        from fast_data_types import VT_PARSER_BUFFER_SIZE
        return VT_PARSER_BUFFER_SIZE - len(self._write_buf)

    def test_commit_write_buffer(self, data, write_buf):
        """Append *data* to the internal write buffer, return bytes consumed."""
        from fast_data_types import VT_PARSER_BUFFER_SIZE
        available = VT_PARSER_BUFFER_SIZE - len(self._write_buf)
        n = min(len(data), available)
        if n > 0:
            self._write_buf.extend(data[:n])
        return n

    def test_parse_written_data(self, cd=None):
        """Parse the write buffer, dispatching to the C terminal and callback."""
        data = bytes(self._write_buf)
        self._write_buf.clear()
        from kitty_parser import parse_bytes
        if cd is not None:
            parse_bytes(self, data, cd)
        elif data:
            # No callback — still parse to keep parser state in sync
            parse_bytes(self, data, lambda *a: None)

    # -- properties --
    @property
    def cursor(self):
        return Cursor(self)

    @property
    def callbacks(self):
        return self._callbacks

    @property
    def columns(self):
        return _lib.kitty_term_cols(self._term)

    @property
    def lines(self):
        return _lib.kitty_term_rows(self._term)

    @property
    def margin_top(self):
        return _read_property(self._ptr, 'scroll_top')

    @property
    def margin_bottom(self):
        return _read_property(self._ptr, 'scroll_bottom')

    @property
    def cursor_key_mode(self):
        return _lib.kitty_term_application_cursor(self._term)

    @cursor_key_mode.setter
    def cursor_key_mode(self, value):
        if value:
            self.write(b'\x1b[?1h')
        else:
            self.write(b'\x1b[?1l')

    # -- line access --
    def line(self, n):
        return Line(self, n)

    # -- operations --
    def reset(self):
        self.write(b'\x1bc')

    def cursor_move(self, n):
        """Move cursor back n cells (kitty cursor_move: direction -1,
        allowed to move to previous lines)."""
        self.write(b'\x1b[' + str(n).encode() + b'D')

    def close(self):
        """Destroy the underlying C terminal."""
        if getattr(self, '_term', None):
            _lib.kitty_term_free(self._term)
            self._term = None

    def __del__(self):
        self.close()