"""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)