Add mouse events to cui.py

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)

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}")