Implement parsing of OSC codes

kitty/boss.py

@@ -16,10 +16,11 @@ from queue import Queue, Empty
 import glfw
 from pyte.streams import Stream, DebugStream
 
+from .constants import appname
 from .char_grid import CharGrid
 from .keys import interpret_text_event, interpret_key_event
 from .screen import Screen
-from .utils import resize_pty, create_pty
+from .utils import resize_pty, create_pty, sanitize_title
 from .fast_data_types import BRACKETED_PASTE_START, BRACKETED_PASTE_END
 
 
@@ -144,7 +145,8 @@ class Boss(Thread):
 
     def render(self):
         if self.pending_title_change is not None:
-            glfw.glfwSetWindowTitle(self.window, self.pending_title_change)
+            t = sanitize_title(self.pending_title_change or appname)
+            glfw.glfwSetWindowTitle(self.window, t)
             self.pending_title_change = None
         if self.pending_icon_change is not None:
             self.pending_icon_change = None  # TODO: Implement this

kitty/data-types.h

@@ -332,6 +332,8 @@ void screen_delete_lines(Screen *self, unsigned int count/*=1*/);
 void screen_delete_characters(Screen *self, unsigned int count);
 void screen_erase_characters(Screen *self, unsigned int count);
 void screen_set_margins(Screen *self, unsigned int top, unsigned int bottom);
+void set_title(Screen *self, const uint8_t *buf, unsigned int sz);
+void set_icon(Screen *self, const uint8_t *buf, unsigned int sz);
 void report_device_attributes(Screen *self, unsigned int UNUSED mode, bool UNUSED secondary);
 void select_graphic_rendition(Screen *self, unsigned int *params, unsigned int count);
 void report_device_status(Screen *self, unsigned int which, bool UNUSED);

kitty/parser.c

@@ -414,8 +414,73 @@ HANDLER(csi) {
 // }}}
 
 // Parse OSC {{{
+
+static inline void handle_osc(Screen *screen, PyObject UNUSED *dump_callback) {
+    unsigned int code = 0;
+    unsigned int start = screen->parser_buf[0] ? screen->parser_buf[0] : 2;
+    unsigned int sz = screen->parser_buf_pos > start ? screen->parser_buf_pos - start : 0;
+    screen->parser_buf[screen->parser_buf_pos] = 0;
+    if (screen->parser_buf[0] && screen->parser_buf[1]) code = (unsigned int)atoi((const char*)screen->parser_buf + 2);
+#define DISPATCH_OSC(name) \
+    REPORT_COMMAND(name, sz); \
+    name(screen, screen->parser_buf + start, sz);
+
+    switch(code) {
+        case 0:
+            DISPATCH_OSC(set_title);
+            DISPATCH_OSC(set_icon);
+            break;
+        case 1:
+            DISPATCH_OSC(set_icon);
+            break;
+        case 2:
+            DISPATCH_OSC(set_title);
+            break;
+        default:
+            REPORT_ERROR("Unknown OSC code: %u", code);
+    }
+#undef DISPATCH_OSC
+}
+
+
 HANDLER(osc) {
-    screen->parser_state = NORMAL_STATE;
+#ifdef DUMP_COMMANDS
+#define HANDLE_OSC handle_osc(screen, dump_callback);
+#else
+#define HANDLE_OSC handle_osc(screen, NULL);
+#endif
+    uint8_t ch = buf[(*pos)++];
+    if (screen->parser_buf_pos == 0) {
+        screen->parser_buf[0] = 0;
+        screen->parser_buf[1] = 1;
+        screen->parser_buf_pos = 2;
+    }
+    switch(ch) {
+        case ST:
+        case BEL:
+            HANDLE_OSC;
+            SET_STATE(NORMAL_STATE);
+            break;
+        case 0:
+            break;  // ignore null bytes
+        case ';':
+            if (!screen->parser_buf[0] && screen->parser_buf_pos < 10) { 
+                // Initial numeric parameter found
+                screen->parser_buf[0] = screen->parser_buf_pos; 
+                break; 
+            }
+        default:
+            if (!screen->parser_buf[0] && (ch < '0' || ch > '9')) {
+                screen->parser_buf[1] = 0;  // No initial numeric parameter
+            }
+            if (screen->parser_buf_pos >= PARSER_BUF_SZ - 1) {
+                REPORT_ERROR("OSC control sequence too long, truncating");
+                HANDLE_OSC;
+                SET_STATE(NORMAL_STATE);
+                break;
+            }
+            screen->parser_buf[screen->parser_buf_pos++] = ch;
+    }
 }
 // }}}
 

kitty/screen.c

@@ -741,16 +741,16 @@ void screen_erase_characters(Screen *self, unsigned int count) {
 
 // Device control {{{
 
-static inline void write_to_child(Screen *self, const char *data, unsigned int sz) {
+static inline void callback(const char *name, Screen *self, const char *data, unsigned int sz) {
-    if (sz) PyObject_CallMethod(self->callbacks, "write_to_child", "y#", data, sz);
+    if (sz) PyObject_CallMethod(self->callbacks, name, "y#", data, sz);
-    else PyObject_CallMethod(self->callbacks, "write_to_child", "y", data);
+    else PyObject_CallMethod(self->callbacks, name, "y", data);
     if (PyErr_Occurred()) PyErr_Print();
     PyErr_Clear(); 
 }
 
 void report_device_attributes(Screen *self, unsigned int UNUSED mode, bool UNUSED secondary) {
     // Do the same as libvte, which gives the below response regardless of mode and secondary
-    write_to_child(self, "\x1b[?62c", 0);  // Corresponds to VT-220
+    callback("write_to_child", self, "\x1b[?62c", 0);  // Corresponds to VT-220
 }
 
 void report_device_status(Screen *self, unsigned int which, bool UNUSED private) {
@@ -760,7 +760,7 @@ void report_device_status(Screen *self, unsigned int which, bool UNUSED private)
     char buf[50] = {0};
     switch(which) {
         case 5:  // device status
-            write_to_child(self, "\x1b[0n", 0); 
+            callback("write_to_child", self, "\x1b[0n", 0); 
             break;
         case 6:  // cursor position
             x = self->cursor->x; y = self->cursor->y;
@@ -770,7 +770,7 @@ void report_device_status(Screen *self, unsigned int which, bool UNUSED private)
             }
             if (self->modes.mDECOM) y -= MAX(y, self->margin_top);
             x++; y++;  // 1-based indexing
-            if (snprintf(buf, sizeof(buf) - 1, "\x1b[%u;%uR", y, x) > 0) write_to_child(self, buf, 0);
+            if (snprintf(buf, sizeof(buf) - 1, "\x1b[%u;%uR", y, x) > 0) callback("write_to_child", self, buf, 0);
             break;
     }
 }
@@ -813,6 +813,14 @@ void screen_set_cursor(Screen *self, unsigned int mode, uint8_t secondary) {
     }
 }
 
+void set_title(Screen *self, const uint8_t *buf, unsigned int sz) {
+    callback("title_changed", self, (const char*)buf, sz);
+}
+
+void set_icon(Screen *self, const uint8_t *buf, unsigned int sz) {
+    callback("icon_changed", self, (const char*)buf, sz);
+}
+
 // }}}
 
 // Python interface {{{

kitty/utils.py

@@ -113,6 +113,8 @@ def timeit(name, do_timing=False):
 
 
 def sanitize_title(x):
+    if isinstance(x, bytes):
+        x = x.decode('utf-8', 'replace')
     return re.sub(r'\s+', ' ', re.sub(r'[\0-\x19]', '', x))
 
 

kitty_tests/parser.py

@@ -25,8 +25,14 @@ class Callbacks:
     def write_to_child(self, data):
         self.wtcbuf += data
 
+    def title_changed(self, data):
+        self.titlebuf += data
+
+    def icon_changed(self, data):
+        self.iconbuf += data
+
     def clear(self):
-        self.wtcbuf = b''
+        self.wtcbuf = self.iconbuf = self.titlebuf = b''
 
 
 class TestScreen(BaseTest):
@@ -123,3 +129,20 @@ class TestScreen(BaseTest):
         pb('\033[1 q', ('screen_set_cursor', 1, ord(' ')))
         self.assertTrue(s.cursor.blink)
         self.ae(s.cursor.shape, CURSOR_BLOCK)
+
+    def test_osc_codes(self):
+        s = self.create_screen()
+        pb = partial(self.parse_bytes_dump, s)
+        c = Callbacks()
+        s.callbacks = c
+        pb(b'a\033]2;xyz\x9cbcde', ('set_title', 3))
+        self.ae(str(s.line(0)), 'abcde')
+        self.ae(c.titlebuf, b'xyz')
+        c.clear()
+        pb('\033]\x07', ('set_title', 0), ('set_icon', 0))
+        self.ae(c.titlebuf, b''), self.ae(c.iconbuf, b'')
+        pb('\033]23\x07', ('set_title', 2), ('set_icon', 2))
+        self.ae(c.titlebuf, b'23'), self.ae(c.iconbuf, b'23')
+        c.clear()
+        pb('\033]2;;;;\x07', ('set_title', 3))
+        self.ae(c.titlebuf, b';;;')