From ff8f5efabd6e2dd52e8c65e308489ac691ecde2c Mon Sep 17 00:00:00 2001 From: GitHub Copilot Date: Sat, 24 Jan 2026 11:27:33 +0000 Subject: [PATCH] Optimize screenshot updates using pyte dirty tracking - get_screen_state() now returns has_changes flag indicating if screen changed - pyte's dirty set tracks which rows have been modified since last read - Screenshot handler returns cached SVG immediately when no changes detected - Removed _screenshot_last_rendered_activity tracking (replaced by dirty flag) - Added test for dirty flag behavior Bump version to 0.1.16 --- pyproject.toml | 2 +- src/textual_webterm/local_server.py | 29 +++++++------------------ src/textual_webterm/terminal_session.py | 15 +++++++++---- tests/test_local_server_unit.py | 16 +++++++------- tests/test_terminal_session.py | 26 ++++++++++++++++++++++ 5 files changed, 54 insertions(+), 34 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index b2767ec..2f3f73c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "textual-webterm" -version = "0.1.15" +version = "0.1.16" description = "Serve terminal sessions over the web" authors = ["Will McGugan "] license = "MIT" diff --git a/src/textual_webterm/local_server.py b/src/textual_webterm/local_server.py index 34aa7db..c3cb559 100644 --- a/src/textual_webterm/local_server.py +++ b/src/textual_webterm/local_server.py @@ -211,7 +211,6 @@ class LocalServer: self._screenshot_cache_etag: dict[str, str] = {} self._screenshot_locks: dict[str, asyncio.Lock] = {} self._route_last_activity: dict[str, float] = {} - self._screenshot_last_rendered_activity: dict[str, float] = {} @property def app_count(self) -> int: @@ -516,30 +515,21 @@ class LocalServer: if session_process is None or not hasattr(session_process, "get_screen_state"): raise web.HTTPNotFound(text="Session not found") - # If nothing has changed since the last render, serve cached screenshot without - # touching the session replay buffer. - last_activity = self._route_last_activity.get(route_key, 0.0) - last_rendered_activity = self._screenshot_last_rendered_activity.get(route_key, -1.0) - if last_activity <= last_rendered_activity: + # Get the actual screen state from the terminal session's pyte screen + # This includes has_changes flag from pyte's dirty tracking + screen_width, screen_height, screen_buffer, has_changes = await session_process.get_screen_state() # type: ignore[union-attr] + + # If screen hasn't changed, serve cached screenshot immediately + cached = self._screenshot_cache.get(route_key) + if cached is not None and not has_changes: cached_response = self._get_cached_screenshot_response(request, route_key) if cached_response is not None: return cached_response - # Get the actual screen state from the terminal session's pyte screen - # This uses the correct dimensions that match what the terminal is rendering - screen_width, screen_height, screen_buffer = await session_process.get_screen_state() # type: ignore[union-attr] - now = asyncio.get_event_loop().time() ttl = self._get_screenshot_cache_ttl(route_key, now) - cached = self._screenshot_cache.get(route_key) - - # If we have a cached screenshot and the session is idle, keep serving it until - # new activity occurs (no periodic re-render). - if cached is not None and self._route_last_activity.get(route_key, 0.0) == 0.0: - cached_response = self._get_cached_screenshot_response(request, route_key) - if cached_response is not None: - return cached_response + # Also check time-based cache for recently rendered screenshots if cached is not None and (now - cached[0]) < ttl: cached_response = self._get_cached_screenshot_response(request, route_key) if cached_response is not None: @@ -620,9 +610,6 @@ class LocalServer: etag = hashlib.sha1(svg.encode("utf-8"), usedforsecurity=False).hexdigest() self._screenshot_cache[route_key] = (asyncio.get_event_loop().time(), svg) self._screenshot_cache_etag[route_key] = etag - self._screenshot_last_rendered_activity[route_key] = self._route_last_activity.get( - route_key, 0.0 - ) headers = {"Cache-Control": "no-cache", "ETag": etag} return web.Response(text=svg, content_type="image/svg+xml", headers=headers) diff --git a/src/textual_webterm/terminal_session.py b/src/textual_webterm/terminal_session.py index 46aac48..760f440 100644 --- a/src/textual_webterm/terminal_session.py +++ b/src/textual_webterm/terminal_session.py @@ -140,16 +140,23 @@ class TerminalSession(Session): async with self._screen_lock: return [line.rstrip() for line in self._screen.display] - async def get_screen_state(self) -> tuple[int, int, list]: + async def get_screen_state(self) -> tuple[int, int, list, bool]: """Get the current screen state including dimensions and character buffer. Returns: - Tuple of (width, height, buffer) where buffer is a list of rows, - each row containing character data with styling attributes. + Tuple of (width, height, buffer, has_changes) where: + - width: screen width in columns + - height: screen height in rows + - buffer: list of rows, each containing character data with styling + - has_changes: True if screen has changed since last call """ async with self._screen_lock: width = self._screen.columns height = self._screen.lines + # Check if any rows are dirty (changed since last clear) + has_changes = len(self._screen.dirty) > 0 + # Clear dirty set after checking + self._screen.dirty.clear() # Copy the buffer data to avoid holding the lock buffer = [] for row in range(height): @@ -166,7 +173,7 @@ class TerminalSession(Session): "reverse": char.reverse, }) buffer.append(row_data) - return (width, height, buffer) + return (width, height, buffer, has_changes) def update_connector(self, connector: SessionConnector) -> None: """Update the connector for reconnection without restarting the session.""" diff --git a/tests/test_local_server_unit.py b/tests/test_local_server_unit.py index 0ba48ca..3a0133e 100644 --- a/tests/test_local_server_unit.py +++ b/tests/test_local_server_unit.py @@ -199,7 +199,7 @@ class TestLocalServerHelpers: [{"data": " ", "fg": "default", "bg": "default", "bold": False, "italics": False, "underscore": False, "reverse": False}] * 80, ] session = MagicMock() - session.get_screen_state = AsyncMock(return_value=(80, 2, screen_buffer)) + session.get_screen_state = AsyncMock(return_value=(80, 2, screen_buffer, True)) monkeypatch.setattr(server.session_manager, "get_session_by_route_key", lambda _rk: session) @@ -222,7 +222,7 @@ class TestLocalServerHelpers: [{"data": " ", "fg": "default", "bg": "default", "bold": False, "italics": False, "underscore": False, "reverse": False}] * 80, ] session = MagicMock() - session.get_screen_state = AsyncMock(return_value=(80, 2, screen_buffer)) + session.get_screen_state = AsyncMock(return_value=(80, 2, screen_buffer, True)) # Pretend app exists for slug "known" server.session_manager.apps_by_slug["known"] = App( @@ -548,23 +548,23 @@ class TestLocalServerMoreCoverage: assert server_with_no_apps._get_screenshot_cache_ttl("rk", now=1000.0) == 60.0 @pytest.mark.asyncio - async def test_handle_screenshot_uses_cache_when_no_new_activity(self, server_with_no_apps, monkeypatch): + async def test_handle_screenshot_uses_cache_when_no_changes(self, server_with_no_apps, monkeypatch): + """Test that cached screenshot is returned when pyte reports no changes.""" request = MagicMock() request.query = {"route_key": "rk"} request.headers = {} + # has_changes=False indicates no screen changes since last call session = MagicMock() - session.get_screen_state = AsyncMock(return_value=(80, 2, [])) + session.get_screen_state = AsyncMock(return_value=(80, 2, [], False)) monkeypatch.setattr(server_with_no_apps.session_manager, "get_session_by_route_key", lambda _rk: session) server_with_no_apps._screenshot_cache["rk"] = (0.0, "cached") server_with_no_apps._screenshot_cache_etag["rk"] = "etag" server_with_no_apps._route_last_activity["rk"] = 5.0 - server_with_no_apps._screenshot_last_rendered_activity["rk"] = 5.0 resp = await server_with_no_apps._handle_screenshot(request) assert "cached" in resp.text - session.get_screen_state.assert_not_awaited() @pytest.mark.asyncio async def test_handle_screenshot_renders_screen_state(self, server_with_no_apps, monkeypatch): @@ -578,7 +578,7 @@ class TestLocalServerMoreCoverage: [{"data": " ", "fg": "default", "bg": "default", "bold": False, "italics": False, "underscore": False, "reverse": False}] * 80, ] session = MagicMock() - session.get_screen_state = AsyncMock(return_value=(80, 2, screen_buffer)) + session.get_screen_state = AsyncMock(return_value=(80, 2, screen_buffer, True)) 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) @@ -760,7 +760,7 @@ class TestLocalServerMoreCoverage: [{"data": c, "fg": "default", "bg": "default", "bold": False, "italics": False, "underscore": False, "reverse": False} for c in "line2" + " " * 75], ] session = MagicMock() - session.get_screen_state = AsyncMock(return_value=(80, 2, screen_buffer)) + session.get_screen_state = AsyncMock(return_value=(80, 2, screen_buffer, True)) 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 diff --git a/tests/test_terminal_session.py b/tests/test_terminal_session.py index be2f355..da6fe54 100644 --- a/tests/test_terminal_session.py +++ b/tests/test_terminal_session.py @@ -130,6 +130,32 @@ class TestTerminalSession: assert lines[1] == "Updated Line 2" assert lines[2] == "Line 3" + @pytest.mark.asyncio + async def test_get_screen_state_returns_dirty_flag(self): + """Test that get_screen_state returns has_changes flag based on pyte dirty tracking.""" + from textual_webterm.terminal_session import TerminalSession + + mock_poller = MagicMock() + session = TerminalSession(mock_poller, "test-session", "bash") + + # After creation, all rows are dirty (initialized) + _w, _h, _buf, has_changes = await session.get_screen_state() + assert has_changes is True # Initial state marks all rows dirty + + # After getting state, dirty set is cleared + # Without new data, has_changes should be False + _, _, _, has_changes = await session.get_screen_state() + assert has_changes is False # No changes since last call + + # Feed new data + await session._update_screen(b"New content\r\n") + _, _, _, has_changes = await session.get_screen_state() + assert has_changes is True # Screen was updated + + # Check again without new data + _, _, _, has_changes = await session.get_screen_state() + assert has_changes is False # No changes + def test_update_connector(self): """Test updating connector.""" from textual_webterm.terminal_session import TerminalSession