[terminal buffering] Address PR feedback: massive simplification of the code

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]] = {
+TERMINAL_MODES_ANSI_SEQUENCES: Mapping[str, str] = {
-    Mode.SynchronizedOutput: {"start_sync": "\x1b[?2026h", "end_sync": "\x1b[?2026l"},
+    "sync_start": "\x1b[?2026h",
+    "sync_stop": "\x1b[?2026l",
 }

src/textual/_terminal_features.py

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

src/textual/_terminal_modes.py

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

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(
+                        if (
-                            self.sender,
+                            mode_report_match["mode_id"] == "2026"
-                            mode_report_match["mode_id"],
+                            and int(mode_report_match["setting_parameter"]) > 0
-                            mode_report_match["setting_parameter"],
+                        ):
-                        )
+                            on_token(
-                        on_token(message)
+                                messages.TerminalSupportsSynchronizedOutput(self.sender)
+                            )
                         break
             else:
                 keys = get_key_ansi_sequence(character, None)

src/textual/app.py

@@ -22,8 +22,7 @@ from typing import (
     TYPE_CHECKING,
 )
 
-from ._terminal_features import TerminalSupportedFeatures
+from ._ansi_sequences import TERMINAL_MODES_ANSI_SEQUENCES
-from ._terminal_modes import Mode, ModeReportResponse
 
 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:
+    def handle_terminal_supports_synchronized_output(
-        if message.mode is Mode.SynchronizedOutput:
+        self, message: messages.TerminalSupportsSynchronizedOutput
-            is_supported = message.mode_is_supported
+    ) -> None:
-            log(
+        log("SynchronizedOutput mode is supported by this terminal")
-                f"SynchronizedOutput mode {'is' if is_supported else 'is not'} supported by this terminal"
+        self._sync_available = True
-            )
-            self._terminal_features = self._terminal_features._replace(
-                synchronised_output=is_supported
-            )
 
     def _begin_update(self) -> None:
-        synchronized_output_start_sequence = (
+        if self._sync_available:
-            self._terminal_features.synchronized_output_start_sequence()
+            self.console.file.write(TERMINAL_MODES_ANSI_SEQUENCES["sync_start"])
-        )
-        if synchronized_output_start_sequence:
-            self.console.file.write(synchronized_output_start_sequence)
 
     def _end_update(self) -> None:
-        synchronized_output_end_sequence = (
+        if self._sync_available:
-            self._terminal_features.synchronized_output_end_sequence()
+            self.console.file.write(TERMINAL_MODES_ANSI_SEQUENCES["sync_stop"])
-        )
-        if synchronized_output_end_sequence:
-            self.console.file.write(synchronized_output_end_sequence)
 
 
 _uvloop_init_done: bool = False

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):
+    def _request_terminal_sync_mode_support(self):
-        self.console.file.write(get_mode_request_sequence(mode))
+        self.console.file.write("\033[?2026$p")
         self.console.file.flush()
 
     @classmethod

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

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
+    """

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][
+_SYNC_START_SEQUENCE = TERMINAL_MODES_ANSI_SEQUENCES["sync_start"]
-    "start_sync"
-]
 
 
 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"