From 43ce3e8363b1a9b9889ea5be6039c702a7ecb230 Mon Sep 17 00:00:00 2001 From: Olivier Philippon Date: Wed, 25 May 2022 12:10:13 +0100 Subject: [PATCH] [terminal buffering] Address PR feedback: massive simplification of the code --- src/textual/_ansi_sequences.py | 10 ++-- src/textual/_terminal_features.py | 42 --------------- src/textual/_terminal_modes.py | 84 ----------------------------- src/textual/_xterm_parser.py | 17 +++--- src/textual/app.py | 34 ++++-------- src/textual/drivers/linux_driver.py | 7 ++- src/textual/events.py | 5 -- src/textual/messages.py | 7 +++ tests/utilities/test_app.py | 12 ++--- 9 files changed, 38 insertions(+), 180 deletions(-) delete mode 100644 src/textual/_terminal_features.py delete mode 100644 src/textual/_terminal_modes.py diff --git a/src/textual/_ansi_sequences.py b/src/textual/_ansi_sequences.py index b467219b3..20b176e2c 100644 --- a/src/textual/_ansi_sequences.py +++ b/src/textual/_ansi_sequences.py @@ -1,10 +1,9 @@ -from typing import Dict, Tuple +from typing import Mapping, Tuple -from ._terminal_modes import Mode from .keys import Keys # Mapping of vt100 escape codes to Keys. -ANSI_SEQUENCES_KEYS: Dict[str, Tuple[Keys, ...]] = { +ANSI_SEQUENCES_KEYS: Mapping[str, Tuple[Keys, ...]] = { # Control keys. "\r": (Keys.Enter,), "\x00": (Keys.ControlAt,), # Control-At (Also for Ctrl-Space) @@ -302,6 +301,7 @@ ANSI_SEQUENCES_KEYS: Dict[str, Tuple[Keys, ...]] = { } -TERMINAL_MODES_ANSI_SEQUENCES: Dict[Mode, Dict[str, str]] = { - Mode.SynchronizedOutput: {"start_sync": "\x1b[?2026h", "end_sync": "\x1b[?2026l"}, +TERMINAL_MODES_ANSI_SEQUENCES: Mapping[str, str] = { + "sync_start": "\x1b[?2026h", + "sync_stop": "\x1b[?2026l", } diff --git a/src/textual/_terminal_features.py b/src/textual/_terminal_features.py deleted file mode 100644 index 4d5e8ddd2..000000000 --- a/src/textual/_terminal_features.py +++ /dev/null @@ -1,42 +0,0 @@ -from __future__ import annotations -from typing import NamedTuple - -from textual._ansi_sequences import TERMINAL_MODES_ANSI_SEQUENCES -from textual._terminal_modes import Mode - - -class TerminalSupportedFeatures(NamedTuple): - """ - Handles information about the features the current terminal emulator seems to support. - """ - - synchronised_output: bool = False - """@link https://gist.github.com/christianparpart/d8a62cc1ab659194337d73e399004036""" - - def synchronized_output_start_sequence(self) -> str: - """ - Returns the ANSI sequence that we should send to the terminal to tell it that - it should start buffering the content we're about to send. - If the terminal doesn't seem to support synchronised updates the string will be empty. - - Returns: - str: the "synchronised output start" ANSI sequence. It will be ab empty string - if the terminal emulator doesn't seem to support the "synchronised updates" mode. - """ - if self.synchronised_output: - return TERMINAL_MODES_ANSI_SEQUENCES[Mode.SynchronizedOutput]["start_sync"] - return "" - - def synchronized_output_end_sequence(self) -> str: - """ - Returns the ANSI sequence that we should send to the terminal to tell it that - it should stop buffering the content we're about to send. - If the terminal doesn't seem to support synchronised updates the string will be empty. - - Returns: - str: the "synchronised output stop" ANSI sequence. It will be ab empty string - if the terminal emulator doesn't seem to support the "synchronised updates" mode. - """ - if self.synchronised_output: - return TERMINAL_MODES_ANSI_SEQUENCES[Mode.SynchronizedOutput]["end_sync"] - return "" diff --git a/src/textual/_terminal_modes.py b/src/textual/_terminal_modes.py deleted file mode 100644 index 601e7543f..000000000 --- a/src/textual/_terminal_modes.py +++ /dev/null @@ -1,84 +0,0 @@ -from __future__ import annotations - -from enum import Enum, IntEnum, unique - -import rich -import rich.repr - -from textual._types import MessageTarget -from textual.message import Message - - -@unique -class Mode(Enum): - # "Modes" are features that terminal emulators can optionally support - # We have to "request" to the emulator if they support a given feature, and they can respond with a "report" - # @link https://vt100.net/docs/vt510-rm/DECRQM.html - # @link https://vt100.net/docs/vt510-rm/DECRPM.html - SynchronizedOutput = "2026" - - -@unique -class ModeReportParameter(IntEnum): - # N.B. The values of this enum are not arbitrary: they match the ones of the spec - # @link https://vt100.net/docs/vt510-rm/DECRPM.html - NotRecognized = 0 - Set = 1 - Reset = 2 - PermanentlySet = 3 - PermanentlyReset = 4 - - -@rich.repr.auto -class ModeReportResponse(Message): - """Sent when the terminals responses to a "mode" (i.e. a supported feature) request""" - - __slots__ = ["mode", "report_parameter"] - - def __init__( - self, sender: MessageTarget, mode: Mode, report_parameter: ModeReportParameter - ) -> None: - """ - Args: - sender (MessageTarget): The sender of the event - mode (Mode): The mode the terminal emulator is giving us results about - report_parameter (ModeReportParameter): The status of this mode for the terminal emulator - """ - super().__init__(sender) - self.mode = mode - self.report_parameter = report_parameter - - @classmethod - def from_terminal_mode_response( - cls, sender: MessageTarget, mode_id: str, setting_parameter: str - ) -> ModeReportResponse: - mode = Mode(mode_id) - report_parameter = ModeReportParameter(int(setting_parameter)) - return ModeReportResponse(sender, mode, report_parameter) - - def __rich_repr__(self) -> rich.repr.Result: - yield "mode", self.mode - yield "report_parameter", self.report_parameter - - @property - def mode_is_supported(self) -> bool: - """Return True if the mode seems to be supported by the terminal emulator. - - Returns: - bool: True if it's supported. False otherwise. - """ - return self.report_parameter in MODE_REPORTS_PARAMETERS_INDICATING_SUPPORT - - -MODE_REPORTS_PARAMETERS_INDICATING_SUPPORT = frozenset( - { - ModeReportParameter.Set, - ModeReportParameter.Reset, - ModeReportParameter.PermanentlySet, - ModeReportParameter.PermanentlyReset, - } -) - - -def get_mode_request_sequence(mode: Mode) -> str: - return "\033[?" + mode.value + "$p\n" diff --git a/src/textual/_xterm_parser.py b/src/textual/_xterm_parser.py index b7742da00..5eaa45106 100644 --- a/src/textual/_xterm_parser.py +++ b/src/textual/_xterm_parser.py @@ -1,13 +1,11 @@ from __future__ import annotations -import os import re from typing import Any, Callable, Generator, Iterable -from . import log +from . import log, messages from . import events -from ._terminal_modes import ModeReportResponse from ._types import MessageTarget from ._parser import Awaitable, Parser, TokenCallback from ._ansi_sequences import ANSI_SEQUENCES_KEYS @@ -128,12 +126,13 @@ class XTermParser(Parser[events.Event]): # Or a mode report? (i.e. the terminal telling us if it supports a mode we requested) mode_report_match = _re_terminal_mode_response.match(sequence) if mode_report_match is not None: - message = ModeReportResponse.from_terminal_mode_response( - self.sender, - mode_report_match["mode_id"], - mode_report_match["setting_parameter"], - ) - on_token(message) + if ( + mode_report_match["mode_id"] == "2026" + and int(mode_report_match["setting_parameter"]) > 0 + ): + on_token( + messages.TerminalSupportsSynchronizedOutput(self.sender) + ) break else: keys = get_key_ansi_sequence(character, None) diff --git a/src/textual/app.py b/src/textual/app.py index 64512c621..3db9cafc8 100644 --- a/src/textual/app.py +++ b/src/textual/app.py @@ -22,8 +22,7 @@ from typing import ( TYPE_CHECKING, ) -from ._terminal_features import TerminalSupportedFeatures -from ._terminal_modes import Mode, ModeReportResponse +from ._ansi_sequences import TERMINAL_MODES_ANSI_SEQUENCES if sys.version_info >= (3, 8): from typing import Literal @@ -145,7 +144,7 @@ class App(Generic[ReturnType], DOMNode): self.driver_class = driver_class or self.get_driver_class() self._title = title self._screen_stack: list[Screen] = [] - self._terminal_features = TerminalSupportedFeatures() + self._sync_available = False self.focused: Widget | None = None self.mouse_over: Widget | None = None @@ -698,7 +697,6 @@ class App(Generic[ReturnType], DOMNode): self.log("---") self.log(driver=self.driver_class) self.log(loop=asyncio.get_running_loop()) - self.log(terminal_features=self._terminal_features) self.log(features=self.features) try: @@ -1060,29 +1058,19 @@ class App(Generic[ReturnType], DOMNode): async def handle_styles_updated(self, message: messages.StylesUpdated) -> None: self.stylesheet.update(self, animate=True) - def handle_mode_report_response(self, message: ModeReportResponse) -> None: - if message.mode is Mode.SynchronizedOutput: - is_supported = message.mode_is_supported - log( - f"SynchronizedOutput mode {'is' if is_supported else 'is not'} supported by this terminal" - ) - self._terminal_features = self._terminal_features._replace( - synchronised_output=is_supported - ) + def handle_terminal_supports_synchronized_output( + self, message: messages.TerminalSupportsSynchronizedOutput + ) -> None: + log("SynchronizedOutput mode is supported by this terminal") + self._sync_available = True def _begin_update(self) -> None: - synchronized_output_start_sequence = ( - self._terminal_features.synchronized_output_start_sequence() - ) - if synchronized_output_start_sequence: - self.console.file.write(synchronized_output_start_sequence) + if self._sync_available: + self.console.file.write(TERMINAL_MODES_ANSI_SEQUENCES["sync_start"]) def _end_update(self) -> None: - synchronized_output_end_sequence = ( - self._terminal_features.synchronized_output_end_sequence() - ) - if synchronized_output_end_sequence: - self.console.file.write(synchronized_output_end_sequence) + if self._sync_available: + self.console.file.write(TERMINAL_MODES_ANSI_SEQUENCES["sync_stop"]) _uvloop_init_done: bool = False diff --git a/src/textual/drivers/linux_driver.py b/src/textual/drivers/linux_driver.py index 49284ba38..409e82032 100644 --- a/src/textual/drivers/linux_driver.py +++ b/src/textual/drivers/linux_driver.py @@ -21,7 +21,6 @@ from ..driver import Driver from ..geometry import Size from .._types import MessageTarget from .._xterm_parser import XTermParser -from .._terminal_modes import get_mode_request_sequence, Mode from .._profile import timer @@ -124,10 +123,10 @@ class LinuxDriver(Driver): self._key_thread = Thread(target=self.run_input_thread, args=(loop,)) send_size_event() self._key_thread.start() - self._request_terminal_mode_support(Mode.SynchronizedOutput) + self._request_terminal_sync_mode_support() - def _request_terminal_mode_support(self, mode: Mode): - self.console.file.write(get_mode_request_sequence(mode)) + def _request_terminal_sync_mode_support(self): + self.console.file.write("\033[?2026$p") self.console.file.flush() @classmethod diff --git a/src/textual/events.py b/src/textual/events.py index 4ad4577cc..caef3719b 100644 --- a/src/textual/events.py +++ b/src/textual/events.py @@ -6,11 +6,6 @@ import rich.repr from rich.style import Style from . import log -from ._terminal_modes import ( - Mode, - ModeReportParameter, - MODE_REPORTS_PARAMETERS_INDICATING_SUPPORT, -) from .geometry import Offset, Size from .message import Message from ._types import MessageTarget diff --git a/src/textual/messages.py b/src/textual/messages.py index 12def496f..165b8719b 100644 --- a/src/textual/messages.py +++ b/src/textual/messages.py @@ -58,3 +58,10 @@ class Prompt(Message, system=True): def can_replace(self, message: Message) -> bool: return isinstance(message, StylesUpdated) + + +class TerminalSupportsSynchronizedOutput(Message): + """ + Used to make the App aware that the terminal emulator supports synchronised output. + @link https://gist.github.com/christianparpart/d8a62cc1ab659194337d73e399004036 + """ diff --git a/tests/utilities/test_app.py b/tests/utilities/test_app.py index d319fb693..98dc8ea7c 100644 --- a/tests/utilities/test_app.py +++ b/tests/utilities/test_app.py @@ -16,8 +16,6 @@ from textual._clock import _Clock from textual.app import WINDOWS from textual._context import active_app from textual._ansi_sequences import TERMINAL_MODES_ANSI_SEQUENCES -from textual._terminal_features import TerminalSupportedFeatures -from textual._terminal_modes import Mode from textual.app import App, ComposeResult from textual.driver import Driver from textual.geometry import Size, Region @@ -25,9 +23,7 @@ from textual.geometry import Size, Region # N.B. These classes would better be named TestApp/TestConsole/TestDriver/etc, # but it makes pytest emit warning as it will try to collect them as classes containing test cases :-/ -CLEAR_SCREEN_SEQUENCE = TERMINAL_MODES_ANSI_SEQUENCES[Mode.SynchronizedOutput][ - "start_sync" -] +_SYNC_START_SEQUENCE = TERMINAL_MODES_ANSI_SEQUENCES["sync_start"] class AppTest(App): @@ -50,10 +46,10 @@ class AppTest(App): # Let's disable all features by default self.features = frozenset() - # We need this so the `CLEAR_SCREEN_SEQUENCE` is always sent for a screen refresh, + # We need this so the "start buffeting"` is always sent for a screen refresh, # whatever the environment: # (we use it to slice the output into distinct full screens displays) - self._terminal_features = TerminalSupportedFeatures(synchronised_output=True) + self._sync_available = True self._size = size self._console = ConsoleTest(width=size.width, height=size.height) @@ -196,7 +192,7 @@ class AppTest(App): total_capture = self.total_capture if not total_capture: return None - screen_captures = total_capture.split(CLEAR_SCREEN_SEQUENCE) + screen_captures = total_capture.split(_SYNC_START_SEQUENCE) for single_screen_capture in reversed(screen_captures): if len(single_screen_capture) > 30: # let's return the last occurrence of a screen that seem to be properly "fully-paint"