trio version blessed.Terminal.inkey

main.py

#!/usr/bin/env python3

import os

import trio
from trio.lowlevel import FdStream

from blessed import Terminal
from blessed.keyboard import Keystroke, resolve_sequence


async def inkey(terminal: Terminal, timeout=None, esc_delay=0.35):
    fd = terminal._keyboard_fd
    assert isinstance(fd, int)

    def resolve(text: str):
        return resolve_sequence(text, mapper=terminal._keymap, codes=terminal._keycodes)

    async def getch(stream: FdStream):
        raw = await stream.receive_some(1)
        if raw == b"":
            raise RuntimeError("keyboard stream has been closed.")

        return terminal._keyboard_decoder.decode(raw, final=False)

    async def internal():
        # TODO@haoliang make Terminal async instead of duplicating keyboard stream
        async with FdStream(os.dup(fd)) as stream:
            buf = ""
            stroke = None

            while not stroke:
                buf += await getch(stream)
                stroke = resolve(buf)

            if stroke.code != terminal.KEY_ESCAPE:
                return buf, stroke

            with trio.move_on_after(esc_delay):
                while (
                    stroke.code == terminal.KEY_ESCAPE
                    and buf in terminal._keymap_prefixes
                ):
                    buf += await getch(stream)
                    stroke = resolve(buf)

            # TODO@haoliang take care of terminal._keyboard_buf

            return buf, stroke

    if timeout is None:
        return await internal()

    try:
        with trio.fail_after(timeout):
            return await internal()
    except trio.TooSlowError:
        # TODO@haoliang or just return what we got above?
        return None, Keystroke("")


async def main():
    terminal = Terminal()
    with terminal.raw():
        raw, key = await inkey(terminal, 3, 1)

    print(f"you pressed key: {key.code=} {key.name=} {raw=}")


def sync_main():
    terminal = Terminal()

    with terminal.raw():
        key = terminal.inkey(3, 1)

    print(f"you pressed key: {key.code=} {key.name=}")


if __name__ == "__main__":
    trio.run(main)
    # sync_main()