mirror of
https://github.com/Textualize/textual.git
synced 2025-10-17 02:38:12 +03:00
[terminal buffering] Remove the management of the iTerm2-specific buffering protocol
Since iTerm2 also supports the "mode 2026", we can just rely on that one ✌
This commit is contained in:
@@ -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"},
|
||||
}
|
||||
|
||||
@@ -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 ""
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
Reference in New Issue
Block a user