kitty keytester

bin/kitty_keytester.py

@@ -0,0 +1,196 @@
+from termios import tcgetattr, tcsetattr, TCSADRAIN, TIOCGWINSZ, TCSAFLUSH
+from tty import setcbreak, setraw
+import os
+import sys
+from contextlib import contextmanager
+from select import select
+from time import sleep
+from json import dumps, loads
+from typing import Optional, Tuple, Union
+
+
+class TerminalContext:
+    def __init__(self, fd: int, close_fd=False) -> None:
+        if not os.isatty(fd):
+            raise TypeError('fd is not a terminal')
+        self.close_fd = close_fd
+        self.fd = fd
+        self.is_cbreak = False
+        self.is_raw = False
+        self._initial_attr = self.termios_attributes
+
+    @classmethod
+    def from_cterm(cls):
+        fd = os.open(os.ctermid(), os.O_RDWR)
+        return TerminalContext(fd, True)
+
+    def close(self) -> None:
+        tcsetattr(self.fd, TCSADRAIN, self._initial_attr)
+        if self.close_fd:
+            os.close(self.fd)
+
+    @property
+    def termios_attributes(self) -> list:
+        return tcgetattr(self.fd)
+
+    @property
+    def ttyname(self) -> str:
+        return os.ttyname(self.fd)
+
+    @contextmanager
+    def cbreak_mode(self):
+        """
+        Enter cbreak mode context.
+        """
+        if self.is_cbreak:
+            yield
+            return
+        tattr = self.termios_attributes
+        try:
+            setcbreak(self.fd, TCSADRAIN)
+            self.is_cbreak = True
+            yield
+        finally:
+            tcsetattr(self.fd, TCSADRAIN, tattr)
+            self.is_cbreak = False
+
+    @contextmanager
+    def raw_mode(self):
+        """
+        Enter cbreak mode context.
+        """
+        if self.is_raw:
+            yield
+            return
+        tattr = self.termios_attributes
+        try:
+            setraw(self.fd, TCSADRAIN)
+            self.is_raw = True
+            yield
+        finally:
+            tcsetattr(self.fd, TCSADRAIN, tattr)
+            self.is_raw = False
+
+    @contextmanager
+    def custom_state(self, undo=None):
+        """
+        Enter custom terminal state, that needs to to be undone by ``undo``.
+        Useful, if you want to apply a custom terminal state and
+        have to make sure, that it gets properly reset to previous state.
+        """
+        try:
+            yield
+        finally:
+            if undo:
+                undo()
+
+    def write(self, data: Union[str, bytes]) -> None:
+        """
+        Write string or bytes directly to the terminal.
+        """
+        if isinstance(data, str):
+            data = data.encode('utf-8')
+        sent = os.write(self.fd, data)
+        while sent:
+            data = data[sent:]
+            sent = os.write(self.fd, data)
+
+    def read(self, amount: int = 1024, timeout: Optional[float] = None) -> bytes:
+        """
+        Blocking read from the terminal.
+        If nothing was sent from the terminal within ``timeout``,
+        empty bytes are returned.
+        """
+        can_read, _, _ = select([self.fd], [], [], timeout)
+        if can_read:
+            return os.read(self.fd, amount)
+        return b''
+
+
+@contextmanager
+def cterminal_context():
+    t = TerminalContext.from_cterm()
+    try:
+        yield t
+    finally:
+        t.close()
+
+
+def extract_events(data: list[str]):
+    if len(data) < 5:
+        raise Exception('not enough reports')
+    types = set(data)
+    if len(types) == 1:
+        return {'PRESS': data[0], 'REPEAT': data[0], 'RELEASE': None}
+    if len(types) == 2:
+        last = data.pop()
+        if last != data[0] and len(set(data)) == 1:
+            return {'PRESS': data[0], 'REPEAT': data[0], 'RELEASE': last}
+        raise Exception('weird reports, 2 types')
+    if len(types) == 3:
+        first = data.pop(0)
+        last = data.pop()
+        if first != last and first != data[0] and last != data[0]:
+            return {'PRESS': first, 'REPEAT': data[0], 'RELEASE': last}
+        raise Exception('weird reports, 3 types')
+    raise Exception('more than 3 types')
+
+
+def query(term: TerminalContext, mode: int):
+    with term.custom_state(undo=lambda:term.write('\x1b[<u')):
+        term.write(f'\x1b[>{mode}u')
+        term.read(timeout=.1)
+        print('PRESS (within 5s) and HOLD (for 5s)\r')
+        data: list[bytes] = []
+        cur = term.read(timeout=5)
+        while cur:
+            data.append(cur)
+            cur = term.read(timeout=.5)
+            if len(data) > 5:
+                print('RELEASE\r', end='')
+        try:
+            return extract_events([b.decode('utf-8') for b in data])
+        except Exception as e:
+            print('Error:', e, '\r')
+            print(data, '\r')
+    term.read(timeout=.1)
+
+
+def save(filedata: dict[str, dict[str, str]], filename, entry, mode, events):
+    try:
+        filedata[entry][mode] = events
+    except KeyError:
+        filedata[entry] = {mode: events}
+    with open(filename, 'w') as f:
+        f.write(dumps(filedata, indent=2))
+
+
+def main():
+    mode = 1
+    filedata = {}
+    print(sys.argv)
+    if len(sys.argv) != 3:
+        print('ERROR: not enough arguments')
+        print('Usage: python kitty_keytester.py <mode> <json-file>')
+        return
+    mode = int(sys.argv[1])
+    filename = sys.argv[2]
+    if os.path.exists(filename):
+        with open(filename) as f:
+            filedata = loads(f.read())
+    with cterminal_context() as term:
+        while True:
+            with term.raw_mode():
+                events = query(term, mode)
+            if events:
+                print(events)
+                entry = input('Entry: ')
+                if entry:
+                    save(filedata, filename, entry, str(mode), events)
+            cont = input('Continue? (y)')
+            if cont not in ['y', '']:
+                break
+
+
+if __name__ == '__main__':
+    main()