fix: use pyte+Rich hybrid for colored SVG screenshots
Screenshots now properly preserve terminal colors: 1. Replay buffer provides raw ANSI data with color codes 2. pyte interprets escape sequences for accurate screen state 3. Rich renders the pyte buffer with colors to SVG This gives us both accurate terminal state (no creeping/wrapping) and proper color preservation in screenshots. Bumps version to 0.1.12.
This commit is contained in:
+1
-1
@@ -1,6 +1,6 @@
|
|||||||
[tool.poetry]
|
[tool.poetry]
|
||||||
name = "textual-webterm"
|
name = "textual-webterm"
|
||||||
version = "0.1.11"
|
version = "0.1.12"
|
||||||
description = "Serve terminal sessions over the web"
|
description = "Serve terminal sessions over the web"
|
||||||
authors = ["Will McGugan <will@textualize.io>"]
|
authors = ["Will McGugan <will@textualize.io>"]
|
||||||
license = "MIT"
|
license = "MIT"
|
||||||
|
|||||||
@@ -14,9 +14,11 @@ from pathlib import Path
|
|||||||
from typing import TYPE_CHECKING
|
from typing import TYPE_CHECKING
|
||||||
|
|
||||||
import aiohttp
|
import aiohttp
|
||||||
|
import pyte
|
||||||
from aiohttp import WSMsgType, web
|
from aiohttp import WSMsgType, web
|
||||||
from rich.ansi import AnsiDecoder
|
|
||||||
from rich.console import Console
|
from rich.console import Console
|
||||||
|
from rich.style import Style
|
||||||
|
from rich.text import Text
|
||||||
|
|
||||||
from . import constants
|
from . import constants
|
||||||
from .exit_poller import ExitPoller
|
from .exit_poller import ExitPoller
|
||||||
@@ -469,11 +471,6 @@ class LocalServer:
|
|||||||
if cached_response is not None:
|
if cached_response is not None:
|
||||||
return cached_response
|
return cached_response
|
||||||
|
|
||||||
# Get screen lines directly from the terminal session's pyte screen
|
|
||||||
# This provides accurate terminal state without replay buffer truncation issues
|
|
||||||
lines = await session_process.get_screen_lines() # type: ignore[union-attr]
|
|
||||||
screen_text = "\n".join(lines)
|
|
||||||
|
|
||||||
try:
|
try:
|
||||||
width = int(request.query.get("width", "120"))
|
width = int(request.query.get("width", "120"))
|
||||||
except ValueError:
|
except ValueError:
|
||||||
@@ -486,6 +483,18 @@ class LocalServer:
|
|||||||
height = DISCONNECT_RESIZE[1]
|
height = DISCONNECT_RESIZE[1]
|
||||||
height = max(5, min(200, height))
|
height = max(5, min(200, height))
|
||||||
|
|
||||||
|
# Hybrid approach for colored screenshots:
|
||||||
|
# 1. Get screen dimensions from pyte (accurate viewport)
|
||||||
|
# 2. Get raw ANSI from replay buffer for colors
|
||||||
|
# 3. Use pyte to interpret the ANSI and get proper screen state
|
||||||
|
# 4. Render through Rich for SVG with colors
|
||||||
|
replay_data = await session_process.get_replay_buffer() # type: ignore[union-attr]
|
||||||
|
# Limit replay data to prevent excessive processing
|
||||||
|
max_replay = 128 * 1024 # 128KB should be plenty for a screen
|
||||||
|
if len(replay_data) > max_replay:
|
||||||
|
replay_data = replay_data[-max_replay:]
|
||||||
|
ansi_text = replay_data.decode("utf-8", errors="replace")
|
||||||
|
|
||||||
now = asyncio.get_event_loop().time()
|
now = asyncio.get_event_loop().time()
|
||||||
ttl = self._get_screenshot_cache_ttl(route_key, now)
|
ttl = self._get_screenshot_cache_ttl(route_key, now)
|
||||||
cached = self._screenshot_cache.get(route_key)
|
cached = self._screenshot_cache.get(route_key)
|
||||||
@@ -517,10 +526,41 @@ class LocalServer:
|
|||||||
return cached_response
|
return cached_response
|
||||||
|
|
||||||
def _render_svg() -> str:
|
def _render_svg() -> str:
|
||||||
|
# Use pyte to interpret ANSI sequences and get accurate screen state
|
||||||
|
screen = pyte.Screen(width, height)
|
||||||
|
stream = pyte.Stream(screen)
|
||||||
|
stream.feed(ansi_text)
|
||||||
|
|
||||||
|
# Convert pyte screen buffer to Rich Text with colors
|
||||||
console = Console(record=True, width=width, height=height, file=io.StringIO())
|
console = Console(record=True, width=width, height=height, file=io.StringIO())
|
||||||
decoder = AnsiDecoder()
|
|
||||||
for renderable in decoder.decode(screen_text):
|
for row in range(height):
|
||||||
console.print(renderable)
|
line = Text()
|
||||||
|
for col in range(width):
|
||||||
|
char = screen.buffer[row][col]
|
||||||
|
char_data = char.data if char.data else " "
|
||||||
|
|
||||||
|
# Build Rich style from pyte character attributes
|
||||||
|
style_kwargs = {}
|
||||||
|
if char.fg != "default":
|
||||||
|
style_kwargs["color"] = char.fg
|
||||||
|
if char.bg != "default":
|
||||||
|
style_kwargs["bgcolor"] = char.bg
|
||||||
|
if char.bold:
|
||||||
|
style_kwargs["bold"] = True
|
||||||
|
if char.italics:
|
||||||
|
style_kwargs["italic"] = True
|
||||||
|
if char.underscore:
|
||||||
|
style_kwargs["underline"] = True
|
||||||
|
if char.reverse:
|
||||||
|
style_kwargs["reverse"] = True
|
||||||
|
|
||||||
|
if style_kwargs:
|
||||||
|
line.append(char_data, Style(**style_kwargs))
|
||||||
|
else:
|
||||||
|
line.append(char_data)
|
||||||
|
|
||||||
|
console.print(line, end="\n", highlight=False)
|
||||||
|
|
||||||
return console.export_svg(
|
return console.export_svg(
|
||||||
title="textual-webterm",
|
title="textual-webterm",
|
||||||
|
|||||||
@@ -195,6 +195,7 @@ class TestLocalServerHelpers:
|
|||||||
|
|
||||||
session = MagicMock()
|
session = MagicMock()
|
||||||
session.get_screen_lines = AsyncMock(return_value=["hello", ""])
|
session.get_screen_lines = AsyncMock(return_value=["hello", ""])
|
||||||
|
session.get_replay_buffer = AsyncMock(return_value=b"hello\r\n")
|
||||||
|
|
||||||
monkeypatch.setattr(server.session_manager, "get_session_by_route_key", lambda _rk: session)
|
monkeypatch.setattr(server.session_manager, "get_session_by_route_key", lambda _rk: session)
|
||||||
|
|
||||||
@@ -213,6 +214,7 @@ class TestLocalServerHelpers:
|
|||||||
|
|
||||||
session = MagicMock()
|
session = MagicMock()
|
||||||
session.get_screen_lines = AsyncMock(return_value=["world", ""])
|
session.get_screen_lines = AsyncMock(return_value=["world", ""])
|
||||||
|
session.get_replay_buffer = AsyncMock(return_value=b"world\r\n")
|
||||||
|
|
||||||
# Pretend app exists for slug "known"
|
# Pretend app exists for slug "known"
|
||||||
server.session_manager.apps_by_slug["known"] = App(
|
server.session_manager.apps_by_slug["known"] = App(
|
||||||
@@ -564,6 +566,7 @@ class TestLocalServerMoreCoverage:
|
|||||||
|
|
||||||
session = MagicMock()
|
session = MagicMock()
|
||||||
session.get_screen_lines = AsyncMock(return_value=["hello", ""])
|
session.get_screen_lines = AsyncMock(return_value=["hello", ""])
|
||||||
|
session.get_replay_buffer = AsyncMock(return_value=b"hello\n")
|
||||||
monkeypatch.setattr(server_with_no_apps.session_manager, "get_session_by_route_key", lambda _rk: session)
|
monkeypatch.setattr(server_with_no_apps.session_manager, "get_session_by_route_key", lambda _rk: session)
|
||||||
|
|
||||||
resp = await server_with_no_apps._handle_screenshot(request)
|
resp = await server_with_no_apps._handle_screenshot(request)
|
||||||
@@ -735,14 +738,15 @@ class TestLocalServerMoreCoverage:
|
|||||||
assert created is True
|
assert created is True
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
async def test_handle_screenshot_uses_get_screen_lines(self, server_with_no_apps, monkeypatch):
|
async def test_handle_screenshot_uses_replay_buffer_with_pyte(self, server_with_no_apps, monkeypatch):
|
||||||
"""Test that screenshot uses get_screen_lines() from terminal session."""
|
"""Test that screenshot uses replay buffer with pyte for colored rendering."""
|
||||||
request = MagicMock()
|
request = MagicMock()
|
||||||
request.query = {"route_key": "rk"}
|
request.query = {"route_key": "rk"}
|
||||||
request.headers = {}
|
request.headers = {}
|
||||||
|
|
||||||
session = MagicMock()
|
session = MagicMock()
|
||||||
session.get_screen_lines = AsyncMock(return_value=["line1", "line2", ""])
|
session.get_screen_lines = AsyncMock(return_value=["line1", "line2", ""])
|
||||||
|
session.get_replay_buffer = AsyncMock(return_value=b"line1\r\nline2\r\n")
|
||||||
monkeypatch.setattr(server_with_no_apps.session_manager, "get_session_by_route_key", lambda _rk: session)
|
monkeypatch.setattr(server_with_no_apps.session_manager, "get_session_by_route_key", lambda _rk: session)
|
||||||
|
|
||||||
server_with_no_apps._route_last_activity["rk"] = 1.0
|
server_with_no_apps._route_last_activity["rk"] = 1.0
|
||||||
@@ -750,4 +754,4 @@ class TestLocalServerMoreCoverage:
|
|||||||
resp = await server_with_no_apps._handle_screenshot(request)
|
resp = await server_with_no_apps._handle_screenshot(request)
|
||||||
assert resp.content_type == "image/svg+xml"
|
assert resp.content_type == "image/svg+xml"
|
||||||
assert "<svg" in resp.text
|
assert "<svg" in resp.text
|
||||||
session.get_screen_lines.assert_awaited_once()
|
session.get_replay_buffer.assert_awaited_once()
|
||||||
|
|||||||
Reference in New Issue
Block a user