From 03fa641be74aed8c6f620a8986ee91ae911cda41 Mon Sep 17 00:00:00 2001 From: Will McGugan Date: Wed, 12 Apr 2023 13:44:22 +0100 Subject: [PATCH] Simplify driver (#2252) * Simplify driver * remove debug flag * added set_terminal_size to driver * restore flush * Restore debug mode * docstring * fix parser * simplify windows driver * driver update * annotations * docstrings --- CHANGELOG.md | 8 +++++--- src/textual/_parser.py | 6 ++++-- src/textual/app.py | 6 ++++-- src/textual/driver.py | 28 +++++++++++++------------- src/textual/drivers/headless_driver.py | 2 +- src/textual/drivers/linux_driver.py | 26 ++++++++++++------------ src/textual/drivers/win32.py | 14 +++++++------ src/textual/drivers/windows_driver.py | 22 +++++++++----------- 8 files changed, 59 insertions(+), 53 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 075ddf5e7..3c55d016b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,15 +7,17 @@ and this project adheres to [Semantic Versioning](http://semver.org/). ## Unreleased +### Changed + +- Changed signature of Driver. Technically a breaking change, but unlikely to affect anyone. +- Breaking change: Timer.start is now private, and returns No + ### Added - Added `DataTable.remove_row` method https://github.com/Textualize/textual/pull/2253 - `Widget.scroll_to_center` now scrolls the widget to the center of the screen https://github.com/Textualize/textual/pull/2255 - Added `TabActivated` message to `TabbedContent` https://github.com/Textualize/textual/pull/2260 -### Changed - -- Breaking change: Timer.start is now private, and returns None ## [0.19.1] - 2023-04-10 diff --git a/src/textual/_parser.py b/src/textual/_parser.py index 54b0f49c6..a1a187da4 100644 --- a/src/textual/_parser.py +++ b/src/textual/_parser.py @@ -114,7 +114,8 @@ class Parser(Generic[T]): _awaiting.remaining = remaining else: _awaiting = self._gen.send(_buffer.getvalue()) - _buffer.truncate(0) + _buffer.seek(0) + _buffer.truncate() elif isinstance(_awaiting, _ReadUntil): chunk = data[pos:] @@ -139,7 +140,8 @@ class Parser(Generic[T]): data = _buffer.getvalue()[sep_index:] pos = 0 self._awaiting = self._gen.send(_buffer.getvalue()[:sep_index]) - _buffer.truncate(0) + _buffer.seek(0) + _buffer.truncate() while tokens: yield popleft() diff --git a/src/textual/app.py b/src/textual/app.py index 15130069f..e37a8dc5d 100644 --- a/src/textual/app.py +++ b/src/textual/app.py @@ -679,7 +679,7 @@ class App(Generic[ReturnType], DOMNode): if driver_import is not None: # The driver class is set from the environment # Syntax should be foo.bar.baz:MyDriver - module_import, colon, driver_symbol = driver_import.partition(":") + module_import, _, driver_symbol = driver_import.partition(":") driver_module = importlib.import_module(module_import) driver_class = getattr(driver_module, driver_symbol) if not inspect.isclass(driver_class) or not issubclass( @@ -1765,7 +1765,9 @@ class App(Generic[ReturnType], DOMNode): HeadlessDriver if headless else self.driver_class, ) driver = self._driver = driver_class( - self.console.file, self, size=terminal_size + self, + debug=constants.DEBUG, + size=terminal_size, ) if not self._exit: diff --git a/src/textual/driver.py b/src/textual/driver.py index 5b58caa25..8dc2a3e5c 100644 --- a/src/textual/driver.py +++ b/src/textual/driver.py @@ -2,20 +2,21 @@ from __future__ import annotations import asyncio from abc import ABC, abstractmethod -from typing import IO +from typing import TYPE_CHECKING from . import _time, events -from ._types import MessageTarget from .events import MouseUp +if TYPE_CHECKING: + from .app import App + class Driver(ABC): """A base class for drivers.""" def __init__( self, - file: IO[str], - target: "MessageTarget", + app: App, *, debug: bool = False, size: tuple[int, int] | None = None, @@ -23,13 +24,12 @@ class Driver(ABC): """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. + app: The App instance. + debug: Enable debug mode. size: Initial size of the terminal or `None` to detect. """ - self._file = file - self._target = target + self._file = app.console.file + self._app = app self._debug = debug self._size = size self._loop = asyncio.get_running_loop() @@ -49,7 +49,7 @@ class Driver(ABC): event: An event. """ asyncio.run_coroutine_threadsafe( - self._target._post_message(event), loop=self._loop + self._app._post_message(event), loop=self._loop ) def process_event(self, event: events.Event) -> None: @@ -58,7 +58,7 @@ class Driver(ABC): Args: event: An event to send. """ - event._set_sender(self._target) + event._set_sender(self._app) if isinstance(event, events.MouseDown): self._mouse_down_time = event.time if event.button: @@ -106,9 +106,6 @@ class Driver(ABC): 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. @@ -117,6 +114,9 @@ class Driver(ABC): data: Raw data. """ + def flush(self) -> None: + """Flush any buffered data.""" + @abstractmethod def start_application_mode(self) -> None: """Start application mode.""" diff --git a/src/textual/drivers/headless_driver.py b/src/textual/drivers/headless_driver.py index 074ccbc57..695a0c196 100644 --- a/src/textual/drivers/headless_driver.py +++ b/src/textual/drivers/headless_driver.py @@ -52,7 +52,7 @@ class HeadlessDriver(Driver): textual_size = Size(width, height) event = events.Resize(textual_size, textual_size) asyncio.run_coroutine_threadsafe( - self._target._post_message(event), + self._app._post_message(event), loop=loop, ) diff --git a/src/textual/drivers/linux_driver.py b/src/textual/drivers/linux_driver.py index 5b7881071..fcdaae2a0 100644 --- a/src/textual/drivers/linux_driver.py +++ b/src/textual/drivers/linux_driver.py @@ -9,46 +9,46 @@ import termios import tty from codecs import getincrementaldecoder from threading import Event, Thread -from typing import IO, Any +from typing import TYPE_CHECKING, Any import rich.repr from .. import events, log -from .._types import MessageTarget from .._xterm_parser import XTermParser from ..driver import Driver from ..geometry import Size +if TYPE_CHECKING: + from ..app import App -@rich.repr.auto + +@rich.repr.auto(angular=True) class LinuxDriver(Driver): """Powers display and input for Linux / MacOS""" def __init__( self, - file: IO[str], - target: "MessageTarget", + app: App, *, debug: bool = False, size: tuple[int, int] | None = None, ) -> None: - """Initialize a driver. + """Initialize Linux driver. Args: - file: A file-like object open for writing unicode. - target: The message target (expected to be the app). - debug: Enabled debug mode. + app: The App instance. + debug: Enable debug mode. size: Initial size of the terminal or `None` to detect. """ - super().__init__(file, target, debug=debug, size=size) - self._file = file + super().__init__(app, debug=debug, size=size) + self._file = app.console.file self.fileno = sys.stdin.fileno() self.attrs_before: list[Any] | None = None self.exit_event = Event() self._key_thread: Thread | None = None def __rich_repr__(self) -> rich.repr.Result: - yield "debug", self._debug + yield self._app def _get_terminal_size(self) -> tuple[int, int]: """Detect the terminal size. @@ -120,7 +120,7 @@ class LinuxDriver(Driver): textual_size = Size(width, height) event = events.Resize(textual_size, textual_size) asyncio.run_coroutine_threadsafe( - self._target._post_message(event), + self._app._post_message(event), loop=loop, ) diff --git a/src/textual/drivers/win32.py b/src/textual/drivers/win32.py index d56ea3808..4942bad92 100644 --- a/src/textual/drivers/win32.py +++ b/src/textual/drivers/win32.py @@ -1,3 +1,5 @@ +from __future__ import annotations + import ctypes import msvcrt import sys @@ -5,13 +7,15 @@ import threading from asyncio import AbstractEventLoop, run_coroutine_threadsafe from ctypes import Structure, Union, byref, wintypes from ctypes.wintypes import BOOL, CHAR, DWORD, HANDLE, SHORT, UINT, WCHAR, WORD -from typing import IO, Callable, List, Optional +from typing import IO, TYPE_CHECKING, Callable, List, Optional -from .._types import MessageTarget from .._xterm_parser import XTermParser from ..events import Event, Resize from ..geometry import Size +if TYPE_CHECKING: + from ..app import App + KERNEL32 = ctypes.WinDLL("kernel32", use_last_error=True) # type: ignore # Console input modes @@ -210,14 +214,12 @@ class EventMonitor(threading.Thread): def __init__( self, loop: AbstractEventLoop, - app, - target: MessageTarget, + app: App, exit_event: threading.Event, process_event: Callable[[Event], None], ) -> None: self.loop = loop self.app = app - self.target = target self.exit_event = exit_event self.process_event = process_event super().__init__() @@ -289,4 +291,4 @@ class EventMonitor(threading.Thread): """Called when terminal size changes.""" size = Size(width, height) event = Resize(size, size) - run_coroutine_threadsafe(self.target._post_message(event), loop=self.loop) + run_coroutine_threadsafe(self.app._post_message(event), loop=self.loop) diff --git a/src/textual/drivers/windows_driver.py b/src/textual/drivers/windows_driver.py index 319e7b3de..369de8245 100644 --- a/src/textual/drivers/windows_driver.py +++ b/src/textual/drivers/windows_driver.py @@ -2,34 +2,34 @@ from __future__ import annotations import asyncio from threading import Event, Thread -from typing import IO, Callable +from typing import TYPE_CHECKING, Callable from .._context import active_app -from .._types import MessageTarget from ..driver import Driver from . import win32 +if TYPE_CHECKING: + from ..app import App + class WindowsDriver(Driver): """Powers display and input for Windows.""" def __init__( self, - file: IO[str], - target: "MessageTarget", + app: App, *, debug: bool = False, size: tuple[int, int] | None = None, ) -> None: - """Initialize a driver. + """Initialize Windows driver. Args: - file: A file-like object open for writing unicode. - target: The message target (expected to be the app). - debug: Enabled debug mode. + app: The App instance. + debug: Enable debug mode. size: Initial size of the terminal or `None` to detect. """ - super().__init__(file, target, debug=debug, size=size) + super().__init__(app, debug=debug, size=size) self.exit_event = Event() self._event_thread: Thread | None = None @@ -81,10 +81,8 @@ class WindowsDriver(Driver): self.write("\033[?1003h\n") self._enable_bracketed_paste() - app = active_app.get() - self._event_thread = win32.EventMonitor( - loop, app, self._target, self.exit_event, self.process_event + loop, self._app, self.exit_event, self.process_event ) self._event_thread.start()