I made a minimal TUI (Text User Interface) program, which I’m planning to bolt onto another program. I’m fairly happy with it, but keyboardescape doesn’t shut it down properly. Pressing control-C three times usually does.
Any advice welcome.
import asyncio
from concurrent.futures import ThreadPoolExecutor
import contextlib
import curses
from dataclasses import dataclass, field
from datetime import datetime
@dataclass
class Screen:
stdscr: curses.window
line2: str = "line 2"
line3: str = "line 3"
input: str = ""
needs_redraw: asyncio.Event = field(default_factory=asyncio.Event)
stop_event: asyncio.Event = field(default_factory=asyncio.Event)
@property
def line1(self) -> str:
return f"Time: {datetime.now().strftime('%H:%M:%S')}"
def change_line2(self, text: str) -> None:
self.line2 = text
self.needs_redraw.set()
def change_line3(self, text: str) -> None:
self.line3 = text
self.needs_redraw.set()
def draw(self) -> None:
self.stdscr.clear()
self.stdscr.addstr(0, 0, self.line1)
self.stdscr.addstr(1, 0, self.line2)
self.stdscr.addstr(2, 0, self.line3)
self.stdscr.addstr(3, 0, f"> {self.input}")
self.stdscr.move(3, len(self.input) + 2)
self.stdscr.refresh()
self.needs_redraw.clear()
async def handle_input_loop(self) -> None:
loop = asyncio.get_event_loop()
with ThreadPoolExecutor(max_workers=1) as executor:
while not self.stop_event.is_set():
key = await loop.run_in_executor(executor, self.stdscr.get_wch)
if key == "\n":
self.change_line2(f"Entered: {self.input}")
self.input = ""
elif key == curses.KEY_BACKSPACE:
self.input = self.input[:-1]
elif isinstance(key, str) and key.isprintable():
self.input += key
self.needs_redraw.set()
async def update_time_loop(self) -> None:
while not self.stop_event.is_set():
self.needs_redraw.set()
await asyncio.sleep(1)
async def main_loop(self) -> None:
while not self.stop_event.is_set():
await self.needs_redraw.wait()
self.draw()
def curses_main(stdscr: curses.window) -> None:
flush_input(stdscr)
screen = Screen(stdscr=stdscr)
async def run_all() -> None:
with contextlib.suppress(asyncio.CancelledError):
await asyncio.gather(
screen.update_time_loop(),
screen.handle_input_loop(),
screen.main_loop(),
)
asyncio.run(run_all())
def flush_input(stdscr: curses.window) -> None:
"""Flush input buffer to avoid stray key input."""
stdscr.nodelay(True)
try:
while True:
stdscr.get_wch()
except curses.error:
pass
stdscr.nodelay(False)
if __name__ == "__main__":
curses.wrapper(curses_main)
(Also if there’s something else I’m doing stupidly.)