diff --git a/sample/abstract_display.py b/sample/abstract_display.py index a2de9e8..4f76670 100644 --- a/sample/abstract_display.py +++ b/sample/abstract_display.py @@ -6,9 +6,10 @@ import _curses import asyncio +import curses import logging from abc import ABC, abstractmethod -from curses import ERR, KEY_RESIZE, curs_set +from curses import ERR, KEY_MOUSE, KEY_RESIZE, curs_set, getmouse from context_sample import GeckoAsyncTaskMan @@ -24,6 +25,16 @@ def __init__(self, stdscr: _curses.window) -> None: self.done_event = asyncio.Event() self.queue = asyncio.Queue(5) + # Enable mouse events + curses.mousemask(curses.ALL_MOUSE_EVENTS | curses.REPORT_MOUSE_POSITION) + + # https://stackoverflow.com/questions/56300134/how-to-enable-mouse-movement-events-in-curses#64809709 + print("\033[?1003h") # enable mouse tracking with the XTERM API # noqa: T201 + # https://invisible-island.net/xterm/ctlseqs/ctlseqs.html#h2-Mouse-Tracking + + # Temp waiting to get replaced + self._commands = {} + @abstractmethod def make_display(self) -> None: """Make a display.""" @@ -32,6 +43,17 @@ def make_display(self) -> None: async def handle_char(self, char: int) -> None: """Handle a character.""" + @abstractmethod + async def handle_mouse(self, mouse: tuple[int, int, int, int, int]) -> None: + """Handle the mouse.""" + + def refresh(self) -> None: + """Refesh the display.""" + self._commands = {} + self.stdscr.erase() + self.make_display() + self.stdscr.refresh() + def set_exit(self) -> None: """Indicagte we can exit.""" self.done_event.set() @@ -51,7 +73,9 @@ async def process_input(self) -> None: # Do nothing and let the loop continue without sleeping continue pass elif char == KEY_RESIZE: - self.make_display() + self.refresh() + elif char == KEY_MOUSE: + await self.handle_mouse(getmouse()) else: await self.handle_char(char) self.queue.task_done() @@ -72,9 +96,34 @@ async def run(self, taskman: GeckoAsyncTaskMan) -> None: curs_set(0) self.stdscr.nodelay(True) # noqa: FBT003 - self.make_display() + self.refresh() taskman.add_task(self.enqueue_input(), "Input gather", "CUI") taskman.add_task(self.process_input(), "Process input", "CUI") await self.done_event.wait() taskman.cancel_key_tasks("CUI") + + def add_text_box( + self, y: int, x: int, text: str | list[str] + ) -> tuple[int, int, int, int]: + """Add a text box.""" + if isinstance(text, str): + text = [text] + width = max([len(t) for t in text]) + height = len(text) + + self.stdscr.addstr(y, x, f"┌{'─' * width}┐") + for idx, line in enumerate(text): + self.stdscr.addstr(y + idx, x, f"│ {line} │") + self.stdscr.addstr(y + height, x, f"└{'─' * width}┘") + return (y, x, height + 2, width + 2) + + def add_button( + self, y: int, z: int, text: str | list[str], char: int | None, click: int + ) -> None: + """Add a button to the window.""" + pos = self.add_text_box(y, z, text) + + def add_line(self, y: int, x: int, text: str) -> None: + """Add a text line.""" + self.stdscr.addstr(y, x, text) diff --git a/sample/cui.py b/sample/cui.py index 5edf57a..2d21c4b 100644 --- a/sample/cui.py +++ b/sample/cui.py @@ -42,17 +42,9 @@ def __init__(self, stdscr: _curses.window) -> None: AbstractDisplay.__init__(self, stdscr) GeckoAsyncSpaMan.__init__(self, CLIENT_ID) - # Enable mouse events - curses.mousemask(1) - self._config = Config() self._spas: GeckoAsyncSpaDescriptor | None = None - self._last_update = time.monotonic() - self._last_char = None - self._commands = {} - self._watching_ping_sensor = False - # Various flags based on the SpaMan events to simulate an # automation client self._can_use_facade = False @@ -74,7 +66,7 @@ async def __aexit__(self, *exc_info: object) -> None: async def _timer_loop(self) -> None: try: while True: - self.make_display() + self.refresh() await config_sleep(1, "CUI Timer") except asyncio.CancelledError: _LOGGER.debug("Timer loop cancelled") @@ -94,7 +86,7 @@ async def handle_event(self, event: GeckoSpaEvent, **_kwargs: Any) -> None: ): self._can_use_facade = False - self.make_display() + self.refresh() async def _select_spa(self, spa: GeckoAsyncSpaDescriptor) -> None: self._config.set_spa_id(spa.identifier_as_string) @@ -144,13 +136,11 @@ def make_display(self) -> None: # noqa: PLR0912, PLR0915 """Make a display.""" try: maxy, maxx = self.stdscr.getmaxyx() - self.stdscr.erase() self.stdscr.box() self.make_title(maxy, maxx) lines = [] - self._commands = {} if self._can_use_facade: assert self.facade is not None # noqa: S101 @@ -268,6 +258,8 @@ def make_display(self) -> None: # noqa: PLR0912, PLR0915 if self._config.spa_id is not None: lines.append("Press 's' to scan for spas") self._commands["s"] = self._clear_spa + lines.append("Press 'f' to flash the screen") + self._commands["f"] = curses.flash lines.append("Press 'q' to exit") self._commands["q"] = self.set_exit @@ -284,12 +276,22 @@ def make_display(self) -> None: # noqa: PLR0912, PLR0915 f"{datetime.now(tz=UTC):%x %X} - {self}", ) + self.stdscr.addstr(1, 2, "+----------+ ╔══════════╗ ┌──────────┐") + self.stdscr.addstr(2, 2, "| A Button | ║ B Button ║ │ C Button │") + self.stdscr.addstr(3, 2, "+----------+ ╚══════════╝ └──────────┘") + + self.add_line(2, 100, "[Line]") + + self.add_text_box(1, 50, "Hello") + self.add_text_box(1, 80, ["One", "Two"]) + except _curses.error: # If window gets too small, we won't output anything _LOGGER.warning("Screen too small") self.stdscr.erase() + self.stdscr.addstr("Window too small") - self.stdscr.refresh() + # self.stdscr.refresh() async def handle_char(self, char: int) -> None: """Handle a command character.""" @@ -306,5 +308,7 @@ async def handle_char(self, char: int) -> None: func(*parms) else: func() - _LOGGER.debug("Back from handling %c", char) - self._last_char = char + + async def handle_mouse(self, mouse: tuple[int, int, int, int, int]) -> None: + """Handle the mouse events.""" + _LOGGER.debug(f"{mouse}")