From 3486dc08b57fd11735bca92fbe325765354d7b4a Mon Sep 17 00:00:00 2001 From: Olivier Philippon Date: Fri, 20 May 2022 16:22:36 +0100 Subject: [PATCH] [terminal buffering] Remove the management of the iTerm2-specific buffering protocol MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Since iTerm2 also supports the "mode 2026", we can just rely on that one ✌ --- src/textual/_ansi_sequences.py | 8 +++++-- src/textual/_terminal_features.py | 34 ++++++++++------------------- src/textual/_terminal_modes.py | 2 +- src/textual/app.py | 4 +++- src/textual/drivers/linux_driver.py | 2 +- tests/utilities/test_app.py | 17 ++++++++------- 6 files changed, 32 insertions(+), 35 deletions(-) diff --git a/src/textual/_ansi_sequences.py b/src/textual/_ansi_sequences.py index a7af17510..86ed71299 100644 --- a/src/textual/_ansi_sequences.py +++ b/src/textual/_ansi_sequences.py @@ -1,7 +1,7 @@ from typing import Dict, Tuple from ._terminal_modes import ( - get__mode_report_sequence, + get_mode_report_sequence, Mode, ModeReportParameter, ) @@ -308,8 +308,12 @@ ANSI_SEQUENCES_KEYS: Dict[str, Tuple[Keys, ...]] = { # Mapping of escape codes to report whether they support a "mode" we requested. ANSI_SEQUENCES_MODE_REPORTS: Dict[str, Tuple[Mode, ModeReportParameter]] = { - get__mode_report_sequence(mode, parameter): (mode, parameter) + get_mode_report_sequence(mode, parameter): (mode, parameter) for mode, parameter in [ (mode, parameter) for parameter in ModeReportParameter for mode in Mode ] } + +TERMINAL_MODES_ANSI_SEQUENCES: Dict[Mode, dict] = { + Mode.SynchronizedOutput: {"start_sync": "\x1b[?2026h", "end_sync": "\x1b[?2026l"}, +} diff --git a/src/textual/_terminal_features.py b/src/textual/_terminal_features.py index 70c28b831..14e4f1be8 100644 --- a/src/textual/_terminal_features.py +++ b/src/textual/_terminal_features.py @@ -1,17 +1,15 @@ from __future__ import annotations -import os -import platform -from dataclasses import dataclass +from typing import NamedTuple + +from textual._ansi_sequences import TERMINAL_MODES_ANSI_SEQUENCES +from textual._terminal_modes import Mode -@dataclass -class TerminalSupportedFeatures: +class TerminalSupportedFeatures(NamedTuple): """ Handles information about the features the current terminal emulator seems to support. """ - iterm2_synchronized_update: bool = False - """@link https://gitlab.com/gnachman/iterm2/-/wikis/synchronized-updates-spec""" mode2026_synchronized_update: bool = False """@link https://gist.github.com/christianparpart/d8a62cc1ab659194337d73e399004036""" @@ -26,19 +24,15 @@ class TerminalSupportedFeatures: TerminalSupportedFeatures: a new TerminalSupportedFeatures """ - # Using macOS, but not using the default terminal: let's assume we're on iTerm2 - iterm2_synchronized_update = ( - platform.system() == "Darwin" - and os.environ.get("TERM_PROGRAM", "") != "Apple_Terminal" - ) + # Here we might detect some features later on, by checking the OS type, the env vars, etc. + # (the same way we were doing it to detect iTerm2 "synchronized update" mode) - # Detecting "mode2026" is more complicated, as we have to use an async request/response + # Detecting "mode2026" is complicated, as we have to use an async request/response # machinery with the terminal emulator - for now we should just assume it's not supported. # See the use of the Mode and ModeReportParameter classes in the Textual code to check this machinery. mode2026_synchronized_update = False return cls( - iterm2_synchronized_update=iterm2_synchronized_update, mode2026_synchronized_update=mode2026_synchronized_update, ) @@ -47,12 +41,12 @@ class TerminalSupportedFeatures: """ Tells the caller if the current terminal emulator seems to support "synchronised updates" (i.e. "buffered" updates). - At the moment we support the iTerm2 specific one, as wel las the more generic "mode 2026". + At the moment we only support the generic "mode 2026". Returns: bool: whether the terminal seems to support buffered mode or not """ - return self.iterm2_synchronized_update or self.mode2026_synchronized_update + return self.mode2026_synchronized_update def synchronized_update_sequences(self) -> tuple[str, str]: """ @@ -70,15 +64,11 @@ class TerminalSupportedFeatures: ) def _synchronized_update_start_sequence(self) -> str: - if self.iterm2_synchronized_update: - return "\x1bP=1s\x1b\\" if self.mode2026_synchronized_update: - return "\x1b[?2026h" + return TERMINAL_MODES_ANSI_SEQUENCES[Mode.SynchronizedOutput]["start_sync"] return "" def _synchronized_update_end_sequence(self) -> str: - if self.iterm2_synchronized_update: - return "\x1bP=2s\x1b\\" if self.mode2026_synchronized_update: - return "\x1b[?2026l" + return TERMINAL_MODES_ANSI_SEQUENCES[Mode.SynchronizedOutput]["end_sync"] return "" diff --git a/src/textual/_terminal_modes.py b/src/textual/_terminal_modes.py index 44e5006e2..fa194a29d 100644 --- a/src/textual/_terminal_modes.py +++ b/src/textual/_terminal_modes.py @@ -37,5 +37,5 @@ def get_mode_request_sequence(mode: Mode) -> str: return "\033[?" + mode.value + "$p\n" -def get__mode_report_sequence(mode: Mode, parameter: ModeReportParameter) -> str: +def get_mode_report_sequence(mode: Mode, parameter: ModeReportParameter) -> str: return f"\x1b[?{mode.value};{parameter.value}$y" diff --git a/src/textual/app.py b/src/textual/app.py index 733bda189..53ede369c 100644 --- a/src/textual/app.py +++ b/src/textual/app.py @@ -952,7 +952,9 @@ class App(Generic[ReturnType], DOMNode): log( f"SynchronizedOutput (aka 'mode2026') {'is' if is_supported else ' is not'} supported" ) - self._terminal_features.mode2026_synchronized_update = is_supported + self._terminal_features = self._terminal_features._replace( + mode2026_synchronized_update=is_supported + ) else: await super().on_event(event) diff --git a/src/textual/drivers/linux_driver.py b/src/textual/drivers/linux_driver.py index 06c40399a..49284ba38 100644 --- a/src/textual/drivers/linux_driver.py +++ b/src/textual/drivers/linux_driver.py @@ -127,7 +127,7 @@ class LinuxDriver(Driver): self._request_terminal_mode_support(Mode.SynchronizedOutput) def _request_terminal_mode_support(self, mode: Mode): - self.console.file.write(get_mode_request_sequence(mode) + "\n") + self.console.file.write(get_mode_request_sequence(mode)) self.console.file.flush() @classmethod diff --git a/tests/utilities/test_app.py b/tests/utilities/test_app.py index a56081a9a..9b6eb88e9 100644 --- a/tests/utilities/test_app.py +++ b/tests/utilities/test_app.py @@ -15,7 +15,9 @@ from textual import events, errors 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 @@ -23,8 +25,9 @@ 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 :-/ -# This value is also hard-coded in Textual's `App` class: -CLEAR_SCREEN_SEQUENCE = "\x1bP=1s\x1b\\" +CLEAR_SCREEN_SEQUENCE = TERMINAL_MODES_ANSI_SEQUENCES[Mode.SynchronizedOutput][ + "start_sync" +] class AppTest(App): @@ -47,12 +50,11 @@ class AppTest(App): # Let's disable all features by default self.features = frozenset() - # We need this so the iTerm2 `CLEAR_SCREEN_SEQUENCE` is always sent for a screen refresh, + # We need this so the `CLEAR_SCREEN_SEQUENCE` 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( - iterm2_synchronized_update=True, - mode2026_synchronized_update=False, + mode2026_synchronized_update=True, ) self._size = size @@ -347,9 +349,8 @@ class ClockMock(_Clock): # ...and let's mark it for removal: activated_events_times_to_clear.append(monotonic_time) - if activated_events_times_to_clear: - for event_time_to_clear in activated_events_times_to_clear: - del self._pending_sleep_events[event_time_to_clear] + for event_time_to_clear in activated_events_times_to_clear: + del self._pending_sleep_events[event_time_to_clear] return activated_timers_count