summaryrefslogtreecommitdiff
path: root/tests/extern/kitty/kitty_parser.py
diff options
context:
space:
mode:
Diffstat (limited to 'tests/extern/kitty/kitty_parser.py')
-rw-r--r--tests/extern/kitty/kitty_parser.py1068
1 files changed, 1068 insertions, 0 deletions
diff --git a/tests/extern/kitty/kitty_parser.py b/tests/extern/kitty/kitty_parser.py
new file mode 100644
index 0000000..f34a608
--- /dev/null
+++ b/tests/extern/kitty/kitty_parser.py
@@ -0,0 +1,1068 @@
+"""Pure-Python VT parser that emits Kitty-style callback events.
+
+This parser mirrors the event vocabulary that Kitty's ``parser.py`` test
+suite expects (``draw``, ``screen_cursor_position``,
+``select_graphic_rendition``, ``set_title``, …) while feeding every byte
+to the C terminal through ``screen.write(data)`` so that observable
+screen state is driven by the real implementation.
+
+The parser maintains state across calls so that partial escape sequences
+split across ``parse_bytes`` invocations are handled correctly — this is
+exercised by ``test_parser_threading``.
+"""
+
+# Parser states
+_GROUND = 0
+_ESC = 1
+_CSI = 2
+_OSC = 3
+_DCS = 4
+_APC = 5
+_PM = 6
+_SOS = 7
+_STR_ESC = 8
+_CHARSET = 9
+_ESC_PCT = 10
+
+# CSI byte classes
+_CSI_PARAM = 0 # 0x30-0x3a (digits, ';', ':')
+_CSI_PREFIX = 1 # 0x3c-0x3f (< = > ?)
+_CSI_INTER = 2 # 0x20-0x2f
+_CSI_FINAL = 3 # 0x40-0x7e
+_CSI_OTHER = 4
+
+_MAX_PARAMS = 256
+
+
+class _CSIState:
+ """CSI parse state, mirroring kitty's vt-parser.c grammar.
+
+ Params accumulate into ``accum`` with ``num_digits`` digits seen;
+ ``commit`` pushes mult*accum. A param with no digits is never
+ committed, so a trailing ';' does not create an empty final param.
+ ``phase``: 0=start, 1=body, 2=post-secondary (after intermediate).
+ """
+
+ __slots__ = ('prefix', 'intermediate', 'final', 'params', 'nparams', 'subparams',
+ 'num_digits', 'mult', 'accum', 'phase')
+
+ def __init__(self):
+ self.reset()
+
+ def reset(self):
+ self.prefix = 0
+ self.intermediate = 0
+ self.final = 0
+ self.params = [0] * _MAX_PARAMS
+ self.nparams = 0
+ self.subparams = 0
+ self.num_digits = 0
+ self.mult = 1
+ self.accum = 0
+ self.phase = 0
+
+ def add_digit(self, d):
+ if self.num_digits >= 17:
+ return
+ self.accum = self.accum * 10 + d
+ self.num_digits += 1
+
+ def commit(self):
+ """Commit the pending param. Returns False on param overflow."""
+ if not self.num_digits:
+ return True
+ if self.nparams >= _MAX_PARAMS:
+ return False
+ v = self.mult * self.accum
+ self.params[self.nparams] = ((v + 2**31) % 2**32) - 2**31
+ self.nparams += 1
+ self.num_digits = 0
+ self.mult = 1
+ self.accum = 0
+ return True
+
+
+class KittyParser:
+ """Stateful VT parser that emits Kitty-style events."""
+
+ def __init__(self):
+ self.state = _GROUND
+ self.text = '' # accumulated UTF-8 text for draw events
+ self.utf8_pending = b'' # incomplete UTF-8 sequence
+ self.csi = _CSIState()
+ self.str_buf = bytearray() # OSC/DCS/APC/PM/SOS payload
+ self.str_type = 0 # which string mode we're in
+ self.charset_slot = 0
+
+ # -- text / draw --
+
+ def _flush_text(self, cb):
+ if self.text:
+ cb(0, 'draw', self.text)
+ self.text = ''
+
+ def _emit(self, cb, *args):
+ self._flush_text(cb)
+ cb(0, *args)
+
+ def _kill_pending(self):
+ """A control byte kills a pending incomplete UTF-8 sequence."""
+ if self.utf8_pending:
+ self.utf8_pending = b''
+ self.text += '\ufffd'
+
+ @staticmethod
+ def _control_event(cb, b):
+ if b == 0x0a:
+ cb(0, 'screen_linefeed')
+ elif b == 0x0d:
+ cb(0, 'screen_carriage_return')
+ elif b == 0x08:
+ cb(0, 'screen_backspace')
+ elif b == 0x09:
+ cb(0, 'screen_tab')
+
+ def _decode_utf8(self, data, cb):
+ """Decode UTF-8 bytes, appending to self.text. Uses U+FFFD for errors."""
+ # Combine pending bytes with new data
+ buf = self.utf8_pending + bytes(data)
+ self.utf8_pending = b''
+ i = 0
+ n = len(buf)
+ while i < n:
+ b = buf[i]
+ # C0 controls (including ESC) terminate text and are handled by caller
+ if b < 0x20:
+ # Do NOT save to utf8_pending — C0 controls are not incomplete UTF-8
+ return i # Return position of the control byte
+ if b < 0x80:
+ self.text += chr(b)
+ i += 1
+ elif b < 0xC2:
+ # Invalid starter (0x80-0xC1) or continuation byte
+ self.text += '\ufffd'
+ i += 1
+ elif b < 0xE0:
+ # 2-byte sequence
+ if i + 1 < n:
+ b2 = buf[i + 1]
+ if 0x80 <= b2 <= 0xBF:
+ self.text += chr((b & 0x1F) << 6 | (b2 & 0x3F))
+ i += 2
+ else:
+ # Bad continuation
+ self.text += '\ufffd'
+ i += 1
+ else:
+ # Incomplete — save for next call
+ self.utf8_pending = buf[i:]
+ return n
+ elif b < 0xF0:
+ # 3-byte sequence
+ if i + 2 < n:
+ b2, b3 = buf[i + 1], buf[i + 2]
+ # Check validity
+ if b == 0xE0:
+ valid = 0xA0 <= b2 <= 0xBF
+ elif b == 0xED:
+ valid = 0x80 <= b2 <= 0x9F
+ else:
+ valid = 0x80 <= b2 <= 0xBF
+ if valid and 0x80 <= b3 <= 0xBF:
+ self.text += chr((b & 0x0F) << 12 | (b2 & 0x3F) << 6 | (b3 & 0x3F))
+ i += 3
+ elif 0x80 <= b2 <= 0xBF:
+ # First continuation is valid, second is not
+ self.text += '\ufffd'
+ i += 2
+ # b3 will be processed next (may be another U+FFFD)
+ else:
+ self.text += '\ufffd'
+ i += 1
+ elif i + 1 < n:
+ b2 = buf[i + 1]
+ if 0x80 <= b2 <= 0xBF:
+ # Could be valid — need more data
+ self.utf8_pending = buf[i:]
+ return n
+ else:
+ self.text += '\ufffd'
+ i += 1
+ else:
+ self.utf8_pending = buf[i:]
+ return n
+ elif b < 0xF5:
+ # 4-byte sequence
+ if i + 3 < n:
+ b2, b3, b4 = buf[i + 1], buf[i + 2], buf[i + 3]
+ if b == 0xF0:
+ valid2 = 0x90 <= b2 <= 0xBF
+ elif b == 0xF4:
+ valid2 = 0x80 <= b2 <= 0x8F
+ else:
+ valid2 = 0x80 <= b2 <= 0xBF
+ if valid2 and 0x80 <= b3 <= 0xBF and 0x80 <= b4 <= 0xBF:
+ cp = (b & 0x07) << 18 | (b2 & 0x3F) << 12 | (b3 & 0x3F) << 6 | (b4 & 0x3F)
+ self.text += chr(cp)
+ i += 4
+ elif 0x80 <= b2 <= 0xBF:
+ if 0x80 <= b3 <= 0xBF:
+ # b4 is bad
+ self.text += '\ufffd'
+ i += 3
+ else:
+ self.text += '\ufffd'
+ i += 2
+ else:
+ self.text += '\ufffd'
+ i += 1
+ else:
+ # Not enough bytes — save for next call
+ self.utf8_pending = buf[i:]
+ return n
+ else:
+ # 0xF5-0xFF: never valid
+ self.text += '\ufffd'
+ i += 1
+ return n
+
+ # -- CSI dispatch --
+
+ def _csi_count(self, idx=0):
+ """Return param[idx] or 1 if zero/missing."""
+ if idx < self.csi.nparams and self.csi.params[idx] > 0:
+ return self.csi.params[idx]
+ return 1
+
+ def _format_sgr(self):
+ """Format SGR params into Kitty-style colon/semicolon groups."""
+ if self.csi.nparams == 0:
+ return ['0']
+ params = self.csi.params[:self.csi.nparams]
+ subparams = self.csi.subparams
+ groups = []
+ current = []
+ i = 0
+ while i < len(params):
+ p = params[i]
+ next_has_colon = (i + 1 < len(params)) and bool(subparams & (1 << (i + 1)))
+ if p in (38, 48, 58):
+ # Extended color — flush current, consume following params
+ if current:
+ groups.append(';'.join(current))
+ current = []
+ sub = [str(p)]
+ i += 1
+ while i < len(params) and params[i] not in (38, 48, 58):
+ sub.append(str(params[i]))
+ i += 1
+ groups.append(':'.join(sub))
+ elif next_has_colon:
+ # This param starts a colon group
+ if current:
+ groups.append(';'.join(current))
+ current = []
+ sub = [str(p)]
+ i += 1
+ while i < len(params) and (subparams & (1 << i)):
+ sub.append(str(params[i]))
+ i += 1
+ groups.append(':'.join(sub))
+ else:
+ current.append(str(p))
+ i += 1
+ if current:
+ groups.append(';'.join(current))
+ return groups
+
+ def _dispatch_csi(self, screen, cb):
+ c = self.csi
+ final = chr(c.intermediate) if c.intermediate else ''
+ # Handle intermediate-based dispatches first
+ if c.intermediate:
+ if c.intermediate == ord(' ') and c.prefix == 0:
+ if c.final == ord('q'):
+ # DECSCUSR: CSI Ps SP q
+ if c.nparams > 0 and c.params[0] in (3, 4):
+ self._emit(cb, 'screen_set_cursor', c.params[0], ord(' '))
+ elif c.nparams > 0 and c.params[0] in (1, 2, 5, 6):
+ self._emit(cb, 'screen_set_cursor', c.params[0], ord(' '))
+ elif c.nparams == 0:
+ self._emit(cb, 'screen_set_cursor', 0, ord(' '))
+ elif c.nparams > 0 and c.params[0] == 3:
+ # Shift codes
+ if c.final == ord('@'):
+ self._emit(cb, 'Shift left escape code not implemented')
+ elif c.final == ord('A'):
+ self._emit(cb, 'Shift right escape code not implemented')
+ elif c.final == ord('S'):
+ self._emit(cb, 'Select presentation directions escape code not implemented')
+ return
+ if c.intermediate == ord('!') and c.prefix == 0:
+ # Soft reset: CSI ! p — no event in Kitty's test
+ return
+ if c.intermediate == ord('$'):
+ if c.final == ord('p'):
+ # DECRQM: CSI ? Ps $ p / CSI Ps $ p
+ mode = c.params[0] if c.nparams > 0 else 0
+ private = 1 if c.prefix == ord('?') else 0
+ self._emit(cb, 'report_mode_status', mode, private)
+ elif c.final == ord('r') and c.prefix == 0:
+ # DECCARA: CSI $ r
+ for event_str in self._format_deccara():
+ self._emit(cb, 'deccara', event_str)
+ return
+ if c.intermediate == ord('*') and c.prefix == 0:
+ # DECSACE: CSI Ps * x
+ if c.nparams > 0:
+ self._emit(cb, 'screen_decsace', c.params[0])
+ else:
+ self._emit(cb, 'screen_decsace', 0)
+ return
+ if c.intermediate == ord('+') and c.final == ord('T'):
+ self._emit(cb, 'screen_reverse_scroll_and_fill_from_scrollback',
+ self._csi_count())
+ return
+ return
+
+ # No intermediate
+ f = c.final
+ pfx = c.prefix
+
+ if f == ord('@'):
+ if c.nparams > 1:
+ self._emit(cb, f'CSI code @ has {c.nparams} > 1 parameters')
+ return
+ if c.nparams == 1 and c.params[0] < 0:
+ self._emit(cb, f'CSI code @ is not allowed to have negative parameter ({c.params[0]})')
+ return
+ self._emit(cb, 'screen_insert_characters', self._csi_count())
+ return
+
+ if f == ord('H') or f == ord('f'):
+ row = c.params[0] if c.nparams > 0 else 1
+ col = c.params[1] if c.nparams > 1 else 1
+ self._emit(cb, 'screen_cursor_position', row, col)
+ return
+
+ if f == ord('J'):
+ mode = c.params[0] if c.nparams > 0 else 0
+ private = 1 if pfx == ord('?') else 0
+ self._emit(cb, 'screen_erase_in_display', mode, private)
+ return
+
+ if f == ord('K'):
+ mode = c.params[0] if c.nparams > 0 else 0
+ private = 1 if pfx == ord('?') else 0
+ self._emit(cb, 'screen_erase_in_line', mode, private)
+ return
+
+ if f == ord('h'):
+ private = 1 if pfx == ord('?') else 0
+ if c.nparams == 0:
+ return
+ for i in range(c.nparams):
+ self._emit(cb, 'screen_set_mode', c.params[i], private)
+ return
+
+ if f == ord('l'):
+ private = 1 if pfx == ord('?') else 0
+ if c.nparams == 0:
+ return
+ for i in range(c.nparams):
+ self._emit(cb, 'screen_reset_mode', c.params[i], private)
+ return
+
+ if f == ord('c'):
+ if pfx == ord('>'):
+ self._emit(cb, 'report_device_attributes', c.params[0] if c.nparams > 0 else 0, 62)
+ elif pfx == ord('='):
+ self._emit(cb, 'report_device_attributes', c.params[0] if c.nparams > 0 else 0, 61)
+ elif pfx == 0:
+ self._emit(cb, 'report_device_attributes', c.params[0] if c.nparams > 0 else 0, 0)
+ return
+
+ if f == ord('m'):
+ for grp in self._format_sgr():
+ self._emit(cb, 'select_graphic_rendition', grp)
+ return
+
+ if f == ord('n'):
+ mode = c.params[0] if c.nparams > 0 else 0
+ private = 1 if pfx == ord('?') else 0
+ self._emit(cb, 'report_device_status', mode, private)
+ return
+
+ if f == ord('r'):
+ if pfx:
+ return
+ if c.nparams == 0:
+ self._emit(cb, 'screen_set_margins', 0, 0)
+ else:
+ top = c.params[0] if c.params[0] > 0 else 1
+ bottom = c.params[1] if c.nparams > 1 and c.params[1] > 0 else 0
+ self._emit(cb, 'screen_set_margins', top, bottom)
+ return
+
+ if f == ord('b'):
+ if c.nparams > 1:
+ return
+ self._emit(cb, 'screen_repeat_character', self._csi_count())
+ return
+
+ if f == ord('T'):
+ if pfx == 0 and c.nparams <= 1:
+ self._emit(cb, 'screen_reverse_scroll', self._csi_count())
+ return
+
+ if f == ord('t'):
+ if pfx == 0 and c.nparams > 0:
+ self._emit(cb, 'screen_report_size', c.params[0], c.params[1] if c.nparams > 1 else 0)
+ return
+
+ if f == ord('p'):
+ return
+
+ # Unhandled CSI — silently ignore (no event)
+
+ def _format_deccara(self):
+ """Format DECCARA params (CSI $ r) into a list of event strings.
+
+ Each event is: 'top;left;bottom;right;sgr_group'
+ """
+ c = self.csi
+ if c.nparams == 0:
+ return ['0;0;0;0;0']
+ params = list(c.params[:c.nparams])
+ while len(params) < 4:
+ params.append(0)
+ subparams = c.subparams
+ rect = ';'.join(str(p) for p in params[:4])
+ # Format SGR groups from params[4:]
+ sgr_params = params[4:]
+ if not sgr_params:
+ return [f'{rect};0']
+ groups = []
+ current = []
+ i = 0
+ while i < len(sgr_params):
+ p = sgr_params[i]
+ next_has_colon = (i + 1 < len(sgr_params)) and bool(subparams & (1 << (i + 5)))
+ if p in (38, 48, 58):
+ if current:
+ groups.append(';'.join(current))
+ current = []
+ sub = [str(p)]
+ i += 1
+ # Consume following params that are subparams (preceded by ':')
+ while i < len(sgr_params) and (subparams & (1 << (i + 4))):
+ sub.append(str(sgr_params[i]))
+ i += 1
+ groups.append(':'.join(sub))
+ elif next_has_colon:
+ if current:
+ groups.append(';'.join(current))
+ current = []
+ sub = [str(p)]
+ i += 1
+ while i < len(sgr_params) and (subparams & (1 << (i + 4))):
+ sub.append(str(sgr_params[i]))
+ i += 1
+ groups.append(':'.join(sub))
+ else:
+ current.append(str(p))
+ i += 1
+ if current:
+ groups.append(';'.join(current))
+ return [f'{rect};{g}' for g in groups]
+
+ # -- OSC dispatch --
+
+ def _dispatch_osc(self, screen, cb):
+ payload = bytes(self.str_buf).decode('utf-8', errors='replace')
+ # Parse OSC code: leading digits form the code, rest is data.
+ # If there's a ';', it separates code from data.
+ sep = payload.find(';')
+ if sep < 0:
+ # No ';' — leading digits are the code, rest is data
+ i = 0
+ while i < len(payload) and payload[i].isdigit():
+ i += 1
+ if i == 0:
+ # No leading digits — treat as OSC 0 with payload as title
+ code = 0
+ rest = payload
+ elif i == len(payload):
+ # Entire payload is digits — code only, no data
+ code = int(payload)
+ rest = ''
+ else:
+ code = int(payload[:i])
+ rest = payload[i:]
+ else:
+ code_str = payload[:sep]
+ rest = payload[sep + 1:]
+ try:
+ code = int(code_str) if code_str else 0
+ except ValueError:
+ code = -1
+
+ if code == 0:
+ self._emit(cb, 'set_title', rest)
+ self._emit(cb, 'set_icon', rest)
+ elif code == 1:
+ self._emit(cb, 'set_icon', rest)
+ elif code == 2:
+ self._emit(cb, 'set_title', rest)
+ elif code == 8:
+ # Hyperlink
+ if not rest:
+ self._emit(cb, 'set_active_hyperlink', None, None)
+ elif ';' not in rest:
+ self._emit(cb, 'Ignoring malformed OSC 8 code')
+ else:
+ params, uri = rest.split(';', 1)
+ if not uri:
+ uri = None
+ link_id = None
+ if params:
+ for kv in params.split(':'):
+ if kv.startswith('id=') and link_id is None:
+ link_id = kv[3:]
+ self._emit(cb, 'set_active_hyperlink', link_id, uri)
+ elif code == 9:
+ self._emit(cb, 'desktop_notify', 9, rest)
+ elif code == 99:
+ self._emit(cb, 'desktop_notify', 99, rest)
+ elif code == 52:
+ self._emit(cb, 'clipboard_control', 52, rest)
+ elif code == 110:
+ self._emit(cb, 'set_dynamic_color', 110, rest)
+ elif code == 22:
+ self._emit(cb, 'set_dynamic_color', 22, rest)
+ else:
+ # Unknown OSC — pass through as set_dynamic_color for some,
+ # or ignore. Kitty's test only uses specific codes.
+ pass
+
+ # -- DCS dispatch --
+
+ def _dispatch_dcs(self, screen, cb):
+ payload = bytes(self.str_buf)
+ if not payload:
+ return
+
+ # DCS +q — request capabilities
+ if payload.startswith(b'+q'):
+ data = payload[2:].decode('ascii', errors='replace')
+ self._emit(cb, 'screen_request_capabilities', 43, data)
+ return
+
+ # DCS $q — request status
+ if payload.startswith(b'$q'):
+ data = payload[2:].decode('ascii', errors='replace')
+ self._emit(cb, 'screen_request_capabilities', ord('$'), data)
+ return
+
+ # DCS @kitty-cmd
+ if payload.startswith(b'@kitty-cmd'):
+ data = payload[len(b'@kitty-cmd'):].decode('utf-8', errors='replace')
+ self._emit(cb, 'handle_remote_cmd', data)
+ return
+
+ # DCS @kitty-print
+ if payload.startswith(b'@kitty-print|'):
+ data = payload[len(b'@kitty-print|'):].decode('ascii', errors='replace')
+ self._emit(cb, 'handle_remote_print', data)
+ return
+
+ # DCS =1s / =2s — pending mode
+ if payload == b'=1s':
+ self._emit(cb, 'screen_start_pending_mode')
+ return
+ if payload == b'=2s':
+ self._emit(cb, 'screen_stop_pending_mode')
+ return
+
+ # -- Graphics command (APC _ G) --
+
+ def _dispatch_graphics(self, screen, cb):
+ payload = bytes(self.str_buf)
+ if not payload:
+ return
+
+ # Short key → full name mapping (from Kitty's graphics protocol)
+ key_map = {
+ 'a': 'action', 'd': 'delete_action', 't': 'transmission_type',
+ 'c': 'compressed', 'f': 'format', 'm': 'more', 'i': 'id',
+ 's': 'data_width', 'o': 'data_offset', 'w': 'width',
+ 'h': 'height', 'x': 'x_offset', 'y': 'y_offset',
+ 'v': 'data_height', 'u': 'data_sz', 'C': 'cursor_movement',
+ 'n': 'num_cells', 'r': 'num_lines', 'X': 'cell_x_offset',
+ 'Y': 'cell_y_offset', 'z': 'z_index', 'p': 'placement_id',
+ 'I': 'image_number', 'q': 'quiet', 'U': 'unicode_placement',
+ 'P': 'parent_id', 'Q': 'parent_placement_id',
+ 'O': 'offset_from_parent_x', 'N': 'offset_from_parent_y',
+ }
+
+ # Parse the graphics command
+ # Format: _G{params};{base64_payload}ESC\
+ # params are key=value pairs separated by commas
+ semi = payload.find(b';')
+ if semi >= 0:
+ cmd_str = payload[:semi]
+ b64_payload = payload[semi + 1:]
+ else:
+ cmd_str = payload
+ b64_payload = b''
+
+ # Parse key=value pairs
+ result = {}
+ # Defaults
+ for f in 'action delete_action transmission_type compressed'.split():
+ result[f] = b'\0'
+ for f in ('format more id data_sz data_offset width height x_offset y_offset '
+ 'data_height data_width cursor_movement num_cells num_lines '
+ 'cell_x_offset cell_y_offset z_index placement_id image_number '
+ 'quiet unicode_placement parent_id parent_placement_id '
+ 'offset_from_parent_x offset_from_parent_y').split():
+ result[f] = 0
+ try:
+ from fast_data_types import base64_decode
+ result[''] = base64_decode(b64_payload).decode('utf-8', errors='replace')
+ except Exception:
+ result[''] = b64_payload.decode('ascii', errors='replace')
+
+ i = 0
+ n = len(cmd_str)
+ while i < n:
+ # Parse key
+ key_start = i
+ while i < n and cmd_str[i] != ord('='):
+ ch = cmd_str[i]
+ if ch == ord(','):
+ self._emit(cb, 'Malformed GraphicsCommand control block, invalid key character: 0x%02x' % ch)
+ return
+ # Only accept known graphics command keys
+ if chr(ch) not in key_map:
+ self._emit(cb, 'Malformed GraphicsCommand control block, invalid key character: 0x%02x' % ch)
+ return
+ i += 1
+ if i >= n:
+ self._emit(cb, 'Malformed GraphicsCommand control block, no = after key')
+ return
+ key = cmd_str[key_start:i].decode('ascii')
+ i += 1 # skip '="
+
+ # Parse value
+ val_start = i
+ full_key = key_map.get(key, key)
+ is_char_key = full_key in ('action', 'delete_action', 'transmission_type', 'compressed')
+ while i < n and cmd_str[i] != ord(','):
+ ch = cmd_str[i]
+ if ch == ord('='):
+ # = after value — check if we have a value yet
+ if i == val_start:
+ # No value parsed yet
+ self._emit(cb, 'Malformed GraphicsCommand control block, expecting an integer value for key: %s' % key)
+ else:
+ self._emit(cb, 'Malformed GraphicsCommand control block, expecting a , or semi-colon after a value, found: 0x%02x' % ch)
+ return
+ if is_char_key:
+ # Character keys accept a single character
+ if i > val_start:
+ # Already got one char — only one expected
+ self._emit(cb, 'Malformed GraphicsCommand control block, expecting a , or semi-colon after a value, found: 0x%02x' % ch)
+ return
+ elif not (ord('0') <= ch <= ord('9') or ch == ord('-')):
+ self._emit(cb, 'Malformed GraphicsCommand control block, expecting an integer value for key: %s' % key)
+ return
+ i += 1
+
+ val_str = cmd_str[val_start:i].decode('ascii')
+ if not val_str:
+ self._emit(cb, 'Malformed GraphicsCommand control block, expecting an integer value')
+ return
+ if is_char_key:
+ result[full_key] = val_str.encode('ascii')
+ else:
+ try:
+ val = int(val_str)
+ except ValueError:
+ self._emit(cb, 'Malformed GraphicsCommand control block, expecting an integer value for key: %s' % key)
+ return
+ if val > 0xFFFFFFFF:
+ self._emit(cb, 'Malformed GraphicsCommand control block, number is too large')
+ return
+ result[full_key] = val
+
+ if i < n and cmd_str[i] == ord(','):
+ i += 1
+
+ self._emit(cb, 'graphics_command', result)
+
+ # -- ESC dispatch --
+
+ def _dispatch_esc(self, ch, cb):
+ if ch == ord('D'):
+ self._emit(cb, 'screen_index')
+ elif ch == ord('E'):
+ self._emit(cb, 'screen_nel')
+ elif ch == ord('c'):
+ self._emit(cb, 'screen_reset')
+ elif ch == ord('M'):
+ self._emit(cb, 'screen_reverse_index')
+ elif ch == ord('7'):
+ self._emit(cb, 'screen_save_cursor')
+ elif ch == ord('8'):
+ self._emit(cb, 'screen_restore_cursor')
+ elif ch == ord('H'):
+ self._emit(cb, 'screen_set_tab')
+ elif ch == ord('['):
+ pass # handled by state transition
+ elif ch == ord(']'):
+ pass # handled by state transition
+ elif ch == ord('P'):
+ pass # handled by state transition
+ elif ch == ord('_'):
+ pass # handled by state transition
+ elif ch == ord('^'):
+ pass # handled by state transition
+ elif ch == ord('X'):
+ pass # handled by state transition
+ elif ch == ord('(') or ch == ord(')'):
+ pass # handled by state transition
+ elif ch == ord('='):
+ self._emit(cb, 'screen_application_keypad', True)
+ elif ch == ord('>'):
+ self._emit(cb, 'screen_application_keypad', False)
+ elif ch == ord('.'):
+ # Unhandled charset escape — needs the next ESC byte
+ pass
+ else:
+ self._emit(cb, 'Unknown char after ESC: 0x%x' % ch)
+
+ # -- Main parse loop --
+
+ def parse(self, screen, data, cb):
+ """Parse *data*, emitting events to *cb* and feeding bytes to screen."""
+ if isinstance(data, str):
+ data = data.encode('utf-8')
+ data = bytes(data)
+
+ # Feed ALL bytes to the C terminal for screen state
+ screen.write(data)
+
+ i = 0
+ n = len(data)
+ while i < n:
+ b = data[i]
+
+ if self.state == _GROUND:
+ if b == 0x1b:
+ self._kill_pending()
+ self._flush_text(cb)
+ self.state = _ESC
+ i += 1
+ elif b < 0x20:
+ # C0 control
+ self._kill_pending()
+ self._flush_text(cb)
+ self._control_event(cb, b)
+ # Other C0 controls: no event
+ i += 1
+ else:
+ # Text byte — decode UTF-8
+ consumed = self._decode_utf8(data[i:], cb)
+ i += consumed
+
+ elif self.state == _ESC:
+ if b == 0x1b:
+ # ESC after ESC
+ self._emit(cb, 'Unknown char after ESC: 0x1b')
+ # Stay in ESC state
+ i += 1
+ elif b < 0x20:
+ # C0 control after ESC — execute it
+ self._flush_text(cb)
+ self._control_event(cb, b)
+ self.state = _GROUND
+ i += 1
+ elif b == ord('['):
+ self.csi.reset()
+ self.state = _CSI
+ i += 1
+ elif b == ord(']'):
+ self.str_buf.clear()
+ self.str_type = _OSC
+ self.state = _OSC
+ i += 1
+ elif b == ord('P'):
+ self.str_buf.clear()
+ self.str_type = _DCS
+ self.state = _DCS
+ i += 1
+ elif b == ord('_'):
+ self.str_buf.clear()
+ self.str_type = _APC
+ self.state = _APC
+ i += 1
+ elif b == ord('^'):
+ self.str_buf.clear()
+ self.str_type = _PM
+ self.state = _PM
+ i += 1
+ elif b == ord('X'):
+ self.str_buf.clear()
+ self.str_type = _SOS
+ self.state = _SOS
+ i += 1
+ elif b == ord('(') or b == ord(')'):
+ self.charset_slot = 0 if b == ord('(') else 1
+ self.state = _CHARSET
+ i += 1
+ elif b == ord('.'):
+ # Unhandled charset escape — needs the next ESC byte
+ # Emit special message when we see the next ESC
+ self.state = _CHARSET
+ self.charset_slot = -1 # special marker
+ i += 1
+ elif b == ord('%'):
+ # ESC % — codepage selector; needs one more byte
+ self.state = _ESC_PCT
+ i += 1
+ else:
+ self._dispatch_esc(b, cb)
+ self.state = _GROUND
+ i += 1
+
+ elif self.state == _ESC_PCT:
+ self.state = _GROUND
+ if b in (ord('G'), ord('@'), ord('/')):
+ # DOCS selectors (ESC % G etc.) — consume silently
+ i += 1
+ else:
+ # Unknown — kitty draws the sequence literally
+ self.text += '%'
+ # Reprocess this byte in GROUND (don't advance)
+
+ elif self.state == _CHARSET:
+ if self.charset_slot == -1:
+ # Unhandled charset escape (ESC .)
+ if b == 0x1b:
+ self._emit(cb, 'Unhandled charset related escape code: 0x2e 0x1b')
+ self.state = _GROUND
+ else:
+ self._emit(cb, 'Unhandled charset related escape code: 0x2e 0x%02x' % b)
+ self.state = _GROUND
+ i += 1
+ else:
+ self._emit(cb, 'screen_designate_charset', self.charset_slot, b)
+ if self.charset_slot == 1:
+ # G1 charset — also emit screen_change_charset
+ self._emit(cb, 'screen_change_charset', 1)
+ self.state = _GROUND
+ i += 1
+
+ elif self.state == _CSI:
+ c = self.csi
+ if b == 0x1b:
+ # ESC interrupts CSI
+ self._flush_text(cb)
+ self.state = _ESC
+ i += 1
+ elif b < 0x20:
+ # C0 control embedded inside CSI — dispatch, keep parsing
+ self._flush_text(cb)
+ self._control_event(cb, b)
+ i += 1
+ elif c.phase == 2:
+ # Post-secondary: only a final byte is valid
+ if 0x40 <= b <= 0x7e:
+ c.final = b
+ self._dispatch_csi(screen, cb)
+ else:
+ self._emit(cb, 'Invalid character in CSI: %s (0x%x), ignoring the sequence'
+ % (chr(b) if 0x21 <= b <= 0x7e else '0x%x' % b, b))
+ self.state = _GROUND
+ i += 1
+ elif c.phase == 0:
+ if b == ord(';'):
+ if c.nparams < _MAX_PARAMS:
+ c.params[c.nparams] = 0
+ c.nparams += 1
+ c.phase = 1
+ elif 0x30 <= b <= 0x39:
+ c.add_digit(b - 0x30)
+ c.phase = 1
+ elif b in (ord('?'), ord('>'), ord('<'), ord('=')):
+ c.prefix = b
+ c.phase = 1
+ elif 0x20 <= b <= 0x2f:
+ if b == ord('-'):
+ c.mult = -1
+ c.num_digits = 1
+ c.phase = 1
+ else:
+ c.intermediate = b
+ c.phase = 2
+ elif 0x40 <= b <= 0x7e:
+ c.final = b
+ self._dispatch_csi(screen, cb)
+ self.state = _GROUND
+ else:
+ self._emit(cb, 'Invalid character in CSI: %s (0x%x), ignoring the sequence'
+ % (chr(b) if 0x21 <= b <= 0x7e else '0x%x' % b, b))
+ self.state = _GROUND
+ i += 1
+ else:
+ # CSI_BODY
+ if 0x30 <= b <= 0x39:
+ c.add_digit(b - 0x30)
+ elif b == ord(';'):
+ if not c.num_digits:
+ c.num_digits = 1 # empty means zero
+ if not c.commit():
+ self._emit(cb, 'CSI escape code has too many parameters, ignoring it')
+ self.state = _GROUND
+ elif b == ord(':'):
+ if c.commit():
+ c.subparams |= 1 << c.nparams
+ else:
+ self._emit(cb, 'CSI escape code has too many parameters, ignoring it')
+ self.state = _GROUND
+ elif 0x20 <= b <= 0x2f:
+ if b == ord('-') and c.num_digits == 0:
+ c.mult = -1
+ c.num_digits = 1
+ elif c.commit():
+ c.intermediate = b
+ c.phase = 2
+ else:
+ self._emit(cb, 'CSI escape code has too many parameters, ignoring it')
+ self.state = _GROUND
+ elif 0x40 <= b <= 0x7e:
+ if c.num_digits == 1 and c.intermediate == 0 and c.mult == -1:
+ # A bare '-' before the final is an intermediate
+ c.num_digits = 0
+ c.intermediate = ord('-')
+ if c.commit():
+ c.final = b
+ self._dispatch_csi(screen, cb)
+ else:
+ self._emit(cb, 'CSI escape code has too many parameters, ignoring it')
+ self.state = _GROUND
+ else:
+ self._emit(cb, 'Invalid character in CSI: %s (0x%x), ignoring the sequence'
+ % (chr(b) if 0x21 <= b <= 0x7e else '0x%x' % b, b))
+ self.state = _GROUND
+ i += 1
+
+ elif self.state in (_OSC, _DCS, _APC, _PM, _SOS):
+ if b == 0x07:
+ # BEL terminates string
+ self._finish_string(screen, cb)
+ self.state = _GROUND
+ i += 1
+ elif b == 0x1b:
+ self.state = _STR_ESC
+ i += 1
+ elif b < 0x20:
+ # C0 control — abort string (except for BEL handled above)
+ self.str_buf.clear()
+ self.state = _GROUND
+ # Process the control byte in GROUND
+ self._control_event(cb, b)
+ i += 1
+ else:
+ self.str_buf.append(b)
+ i += 1
+
+ elif self.state == _STR_ESC:
+ if b == ord('\\'):
+ # ST (ESC \) terminates string
+ self._finish_string(screen, cb)
+ self.state = _GROUND
+ i += 1
+ else:
+ # Other byte after ESC in string — reprocess
+ self.str_buf.clear()
+ self.state = _GROUND
+ # Don't consume — reprocess in GROUND/ESC
+ # Actually, the ESC should be reprocessed
+ self.state = _ESC
+ i += 0 # Don't advance — reprocess this byte via ESC
+ # But we already consumed the ESC, so process this byte in ESC state
+ if b == 0x1b:
+ self._emit(cb, 'Unknown char after ESC: 0x1b')
+ # Stay in ESC
+ elif b < 0x20:
+ self._control_event(cb, b)
+ self.state = _GROUND
+ else:
+ self._dispatch_esc(b, cb)
+ self.state = _GROUND
+ i += 1
+
+ # End of data — flush any pending text
+ self._flush_text(cb)
+
+ def _finish_string(self, screen, cb):
+ """Dispatch the completed string (OSC/DCS/APC/PM/SOS)."""
+ st = self.str_type
+ if st == _OSC:
+ self._dispatch_osc(screen, cb)
+ elif st == _DCS:
+ self._dispatch_dcs(screen, cb)
+ elif st == _APC:
+ # Check for graphics command (_G)
+ payload = bytes(self.str_buf)
+ if payload.startswith(b'G'):
+ self.str_buf = bytearray(payload[1:])
+ self._dispatch_graphics(screen, cb)
+ else:
+ code = payload[0] if payload else 0
+ self._emit(cb, 'Unrecognized APC code: 0x%02x' % code)
+ elif st == _PM:
+ code = self.str_buf[0] if self.str_buf else 0
+ self._emit(cb, 'Unrecognized PM code: 0x%02x' % code)
+ elif st == _SOS:
+ code = self.str_buf[0] if self.str_buf else 0
+ self._emit(cb, 'Unrecognized SOS code: 0x%02x' % code)
+ self.str_buf.clear()
+
+
+# Module-level parse_bytes function
+def parse_bytes(screen, data, callback=None):
+ """Parse *data* and emit events to *callback*.
+
+ Maintains parser state on the screen object so partial sequences
+ split across ``parse_bytes`` invocations are handled correctly.
+ Also updates screen.callbacks for OSC events (titlebuf, etc.).
+ """
+ if callback is None:
+ callback = lambda *args: None
+ if not hasattr(screen, '_kitty_parser'):
+ screen._kitty_parser = KittyParser()
+
+ def wrapped_cb(wid, *args):
+ # Update screen callbacks for OSC/DCS events
+ if args and args[0] == 'set_title':
+ screen.callbacks.titlebuf.append(args[1])
+ elif args and args[0] == 'set_icon':
+ screen.callbacks.iconbuf += args[1] if isinstance(args[1], str) else str(args[1], 'utf-8')
+ elif args and args[0] == 'desktop_notify':
+ screen.callbacks.notifications.append((args[1], args[2]))
+ elif args and args[0] == 'set_dynamic_color':
+ val = args[2] if isinstance(args[2], str) else str(args[2], 'utf-8')
+ screen.callbacks.colorbuf += val
+ elif args and args[0] == 'handle_remote_print':
+ try:
+ from fast_data_types import base64_decode
+ screen.callbacks.printbuf.append(base64_decode(args[1]).decode('utf-8'))
+ except Exception:
+ screen.callbacks.printbuf.append(args[1])
+ # Forward to original callback
+ callback(wid, *args)
+
+ screen._kitty_parser.parse(screen, data, wrapped_cb)
generated by cgit v1.2.3 (git 2.39.1) at 2026-10-25 02:21:11 +0000