alihan/textual
1
0
Fork 0
mirror of https://github.com/Textualize/textual.git synced 2025-10-17 02:38:12 +03:00
Code Issues Packages Projects Releases Wiki Activity

Simplify driver (#2252)

Browse Source 
* 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
This commit is contained in:
Will McGugan
2023-04-12 13:44:22 +01:00
committed by GitHub
parent 71becfc090
commit 03fa641be7
8 changed files with 59 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
CHANGELOG.md
View File

@@ -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

6
src/textual/_parser.py
View File

@@ -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()

6
src/textual/app.py
View File

@@ -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:

28
src/textual/driver.py
View File

@@ -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."""

2
src/textual/drivers/headless_driver.py
View File

@@ -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,
)

26
src/textual/drivers/linux_driver.py
View File

@@ -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,
)

14
src/textual/drivers/win32.py
View File

@@ -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)

22
src/textual/drivers/windows_driver.py
View File

@@ -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()