Simplify driver (#2091)

* simplify driver

* fix headless driver

* docstrings and simplify

* tidy

* docstrings

* docstring

* docstring

* more docstrings

* import

* Update src/textual/app.py

Co-authored-by: Rodrigo Girão Serrão <[email protected]>

* Update src/textual/driver.py

Co-authored-by: Rodrigo Girão Serrão <[email protected]>

* docstring

* Update src/textual/drivers/linux_driver.py

Co-authored-by: Rodrigo Girão Serrão <[email protected]>

* Update src/textual/drivers/linux_driver.py

Co-authored-by: Rodrigo Girão Serrão <[email protected]>

* Update src/textual/drivers/linux_driver.py

Co-authored-by: Rodrigo Girão Serrão <[email protected]>

* docstring

---------

Co-authored-by: Rodrigo Girão Serrão <[email protected]>

Author: willmcgugan

src/textual/app.py

@@ -507,12 +507,12 @@ def animate(
 
     @property
     def debug(self) -> bool:
-        """bool: Is debug mode is enabled?"""
+        """Is debug mode enabled?"""
         return "debug" in self.features
 
     @property
     def is_headless(self) -> bool:
-        """bool: Is the app running in 'headless' mode?"""
+        """Is the driver running in 'headless' mode?"""
         return False if self._driver is None else self._driver.is_headless
 
     @property
@@ -1619,7 +1619,9 @@ async def invoke_ready_callback() -> None:
                 "type[Driver]",
                 HeadlessDriver if headless else self.driver_class,
             )
-            driver = self._driver = driver_class(self.console, self, size=terminal_size)
+            driver = self._driver = driver_class(
+                self.console.file, self, size=terminal_size
+            )
 
             if not self._exit:
                 driver.start_application_mode()

src/textual/driver.py

@@ -2,26 +2,33 @@
 
 import asyncio
 from abc import ABC, abstractmethod
-from typing import TYPE_CHECKING
+from typing import IO
 
 from . import _time, events
 from ._types import MessageTarget
 from .events import MouseUp
 
-if TYPE_CHECKING:
-    from rich.console import Console
-
 
 class Driver(ABC):
+    """A base class for drivers."""
+
     def __init__(
         self,
-        console: "Console",
+        file: IO[str],
         target: "MessageTarget",
         *,
         debug: bool = False,
         size: tuple[int, int] | None = None,
     ) -> None:
-        self.console = console
+        """Initialize a driver.
+
+        Args:
+            file: A file-like object open for writing unicode.
+            target: The message target (expected to be the app).
+            debug: Enabled debug mode.
+            size: Initial size of the terminal or `None` to detect.
+        """
+        self._file = file
         self._target = target
         self._debug = debug
         self._size = size
@@ -32,16 +39,25 @@ def __init__(
 
     @property
     def is_headless(self) -> bool:
-        """Check if the driver is 'headless'"""
+        """Is the driver 'headless' (no output)?"""
         return False
 
     def send_event(self, event: events.Event) -> None:
+        """Send an event to the target app.
+
+        Args:
+            event: An event.
+        """
         asyncio.run_coroutine_threadsafe(
             self._target._post_message(event), loop=self._loop
         )
 
     def process_event(self, event: events.Event) -> None:
-        """Performs some additional processing of events."""
+        """Perform additional processing on an event, prior to sending.
+
+        Args:
+            event: An event to send.
+        """
         event._set_sender(self._target)
         if isinstance(event, events.MouseDown):
             self._mouse_down_time = event.time
@@ -87,14 +103,25 @@ def process_event(self, event: events.Event) -> None:
             click_event = events.Click.from_event(event)
             self.send_event(click_event)
 
+    def flush(self) -> None:
+        """Flush any buffered data."""
+
+    @abstractmethod
+    def write(self, data: str) -> None:
+        """Write data to the output device.
+
+        Args:
+            data: Raw data.
+        """
+
     @abstractmethod
     def start_application_mode(self) -> None:
-        ...
+        """Start application mode."""
 
     @abstractmethod
     def disable_input(self) -> None:
-        ...
+        """Disable further input."""
 
     @abstractmethod
     def stop_application_mode(self) -> None:
-        ...
+        """Stop application mode, restore state."""

src/textual/drivers/headless_driver.py

@@ -12,6 +12,7 @@ class HeadlessDriver(Driver):
 
     @property
     def is_headless(self) -> bool:
+        """Is the driver running in 'headless' mode?"""
         return True
 
     def _get_terminal_size(self) -> tuple[int, int]:
@@ -32,10 +33,20 @@ def _get_terminal_size(self) -> tuple[int, int]:
         height = height or 25
         return width, height
 
+    def write(self, data: str) -> None:
+        """Write data to the output device.
+
+        Args:
+            data: Raw data.
+        """
+        # Nothing to write as this is a headless driver.
+
     def start_application_mode(self) -> None:
+        """Start application mode."""
         loop = asyncio.get_running_loop()
 
-        def send_size_event():
+        def send_size_event() -> None:
+            """Send first resize event."""
             terminal_size = self._get_terminal_size()
             width, height = terminal_size
             textual_size = Size(width, height)
@@ -48,7 +59,8 @@ def send_size_event():
         send_size_event()
 
     def disable_input(self) -> None:
-        pass
+        """Disable further input."""
 
     def stop_application_mode(self) -> None:
-        pass
+        """Stop application mode, restore state."""
+        # Nothing to do

src/textual/drivers/linux_driver.py

@@ -9,10 +9,7 @@
 import tty
 from codecs import getincrementaldecoder
 from threading import Event, Thread
-from typing import TYPE_CHECKING, Any
-
-if TYPE_CHECKING:
-    from rich.console import Console
+from typing import IO, Any
 
 import rich.repr
 
@@ -29,13 +26,22 @@ class LinuxDriver(Driver):
 
     def __init__(
         self,
-        console: "Console",
+        file: IO[str],
         target: "MessageTarget",
         *,
         debug: bool = False,
         size: tuple[int, int] | None = None,
     ) -> None:
-        super().__init__(console, target, debug=debug, size=size)
+        """Initialize a driver.
+
+        Args:
+            file: A file-like object open for writing unicode.
+            target: The message target (expected to be the app).
+            debug: Enabled debug mode.
+            size: Initial size of the terminal or `None` to detect.
+        """
+        super().__init__(file, target, debug=debug, size=size)
+        self._file = file
         self.fileno = sys.stdin.fileno()
         self.attrs_before: list[Any] | None = None
         self.exit_event = Event()
@@ -45,6 +51,11 @@ def __rich_repr__(self) -> rich.repr.Result:
         yield "debug", self._debug
 
     def _get_terminal_size(self) -> tuple[int, int]:
+        """Detect the terminal size.
+
+        Returns:
+            The size of the terminal as a tuple of (WIDTH, HEIGHT).
+        """
         width: int | None = 80
         height: int | None = 25
         import shutil
@@ -61,35 +72,46 @@ def _get_terminal_size(self) -> tuple[int, int]:
         return width, height
 
     def _enable_mouse_support(self) -> None:
-        write = self.console.file.write
+        """Enable reporting of mouse events."""
+        write = self.write
         write("\x1b[?1000h")  # SET_VT200_MOUSE
         write("\x1b[?1003h")  # SET_ANY_EVENT_MOUSE
         write("\x1b[?1015h")  # SET_VT200_HIGHLIGHT_MOUSE
         write("\x1b[?1006h")  # SET_SGR_EXT_MODE_MOUSE
 
         # write("\x1b[?1007h")
-        self.console.file.flush()
+        self.flush()
 
         # Note: E.g. lxterminal understands 1000h, but not the urxvt or sgr
         #       extensions.
 
     def _enable_bracketed_paste(self) -> None:
         """Enable bracketed paste mode."""
-        self.console.file.write("\x1b[?2004h")
+        self.write("\x1b[?2004h")
 
     def _disable_bracketed_paste(self) -> None:
         """Disable bracketed paste mode."""
-        self.console.file.write("\x1b[?2004l")
+        self.write("\x1b[?2004l")
 
     def _disable_mouse_support(self) -> None:
-        write = self.console.file.write
+        """Disable reporting of mouse events."""
+        write = self.write
         write("\x1b[?1000l")  #
         write("\x1b[?1003l")  #
         write("\x1b[?1015l")
         write("\x1b[?1006l")
-        self.console.file.flush()
+        self.flush()
+
+    def write(self, data: str) -> None:
+        """Write data to the output device.
+
+        Args:
+            data: Raw data.
+        """
+        self._file.write(data)
 
     def start_application_mode(self):
+        """Start application mode."""
         loop = asyncio.get_running_loop()
 
         def send_size_event():
@@ -107,7 +129,8 @@ def on_terminal_resize(signum, stack) -> None:
 
         signal.signal(signal.SIGWINCH, on_terminal_resize)
 
-        self.console.set_alt_screen(True)
+        self.write("\x1b[?1049h")  # Alt screen
+
         self._enable_mouse_support()
         try:
             self.attrs_before = termios.tcgetattr(self.fileno)
@@ -132,10 +155,10 @@ def on_terminal_resize(signum, stack) -> None:
 
             termios.tcsetattr(self.fileno, termios.TCSANOW, newattr)
 
-        self.console.show_cursor(False)
-        self.console.file.write("\033[?1003h\n")
-        self.console.file.flush()
-        self._key_thread = Thread(target=self.run_input_thread, args=(loop,))
+        self.write("\x1b[?25l")  # Hide cursor
+        self.write("\033[?1003h\n")
+        self.flush()
+        self._key_thread = Thread(target=self.run_input_thread)
         send_size_event()
         self._key_thread.start()
         self._request_terminal_sync_mode_support()
@@ -146,9 +169,9 @@ def _request_terminal_sync_mode_support(self) -> None:
         # Terminals should ignore this sequence if not supported.
         # Apple terminal doesn't, and writes a single 'p' in to the terminal,
         # so we will make a special case for Apple terminal (which doesn't support sync anyway).
-        if self.console._environ.get("TERM_PROGRAM", "") != "Apple_Terminal":
-            self.console.file.write("\033[?2026$p")
-            self.console.file.flush()
+        if os.environ.get("TERM_PROGRAM", "") != "Apple_Terminal":
+            self.write("\033[?2026$p")
+            self.flush()
 
     @classmethod
     def _patch_lflag(cls, attrs: int) -> int:
@@ -170,6 +193,7 @@ def _patch_iflag(cls, attrs: int) -> int:
         )
 
     def disable_input(self) -> None:
+        """Disable further input."""
         try:
             if not self.exit_event.is_set():
                 signal.signal(signal.SIGWINCH, signal.SIG_DFL)
@@ -184,6 +208,7 @@ def disable_input(self) -> None:
             pass
 
     def stop_application_mode(self) -> None:
+        """Stop application mode, restore state."""
         self._disable_bracketed_paste()
         self.disable_input()
 
@@ -193,17 +218,12 @@ def stop_application_mode(self) -> None:
             except termios.error:
                 pass
 
-        with self.console:
-            self.console.set_alt_screen(False)
-            self.console.show_cursor(True)
-
-    def run_input_thread(self, loop) -> None:
-        try:
-            self._run_input_thread(loop)
-        except Exception:
-            pass  # TODO: log
+            # Alt screen false, show cursor
+            self.write("\x1b[?1049l" + "\x1b[?25h")
+            self.flush()
 
-    def _run_input_thread(self, loop) -> None:
+    def run_input_thread(self) -> None:
+        """Wait for input and dispatch events."""
         selector = selectors.DefaultSelector()
         selector.register(self.fileno, selectors.EVENT_READ)