Telnet/SSHTerminal -> RemoteTerminal

Author: jquast

gambaterm/remote_terminal.py

@@ -0,0 +1,69 @@
+"""
+Provide a blessed Terminal subclass for remote (SSH/telnet) streams.
+"""
+
+from __future__ import annotations
+
+import codecs
+import contextlib
+from typing import IO, Generator
+
+from blessed import Terminal as BlessedTerminal
+from blessed.terminal import WINSZ
+
+# Python's curses.setupterm() can only be called once per process — subsequent
+# calls with a different terminal type are silently ignored. Since the SSH
+# server handles multiple concurrent connections in threads, all RemoteTerminal
+# instances share whatever terminal type was initialized first by the local
+# Terminal(). We hardcode 'xterm-256color' as the kind since:
+#   1. It's universally compatible with modern terminals
+#   2. We use standard VT100/ANSI escape codes directly, not terminfo caps
+#   3. It avoids issues where the first client's TERM value differs from subsequent
+REMOTE_TERMINAL_TYPE = "xterm-256color"
+
+
+class RemoteTerminal(BlessedTerminal):
+    """A blessed Terminal subclass for remote streams (SSH, telnet).
+
+    Stubs raw/cbreak mode (the remote connection is already raw) and
+    overrides size detection to use values provided by the server.
+    """
+
+    def __init__(
+        self,
+        stream: IO[str],
+        keyboard_fd: int,
+        rows: int,
+        columns: int,
+    ) -> None:
+        self._rows = rows
+        self._columns = columns
+        super().__init__(kind=REMOTE_TERMINAL_TYPE, stream=stream, force_styling=True)
+        # Blessed only sets _keyboard_fd when stream is sys.__stdout__, so
+        # for remote pipes we must set it and initialize the decoder manually
+        self._keyboard_fd = keyboard_fd  # type: ignore[assignment]
+        self._keyboard_decoder = codecs.getincrementaldecoder("UTF-8")()
+
+    @property
+    def is_a_tty(self) -> bool:
+        return True
+
+    @contextlib.contextmanager
+    def raw(self) -> Generator[None, None, None]:
+        yield
+
+    @contextlib.contextmanager
+    def cbreak(self) -> Generator[None, None, None]:
+        yield
+
+    def _height_and_width(self) -> WINSZ:
+        return WINSZ(
+            ws_row=self._rows,
+            ws_col=self._columns,
+            ws_xpixel=0,
+            ws_ypixel=0,
+        )
+
+    def update_size(self, rows: int, columns: int) -> None:
+        self._rows = rows
+        self._columns = columns

gambaterm/ssh.py

@@ -27,7 +27,8 @@
 from .main import add_base_arguments, add_optional_arguments, AppConfig
 from .console import Console, GameboyColor
 
-from .ssh_app_session import SSHTerminal, process_to_terminal
+from .remote_terminal import RemoteTerminal
+from .ssh_app_session import process_to_terminal
 
 
 def is_x11_display_functional(
@@ -149,7 +150,7 @@ async def ssh_process_handler(process: SSHServerProcess[str]) -> int:
 
 
 def ssh_terminal_handler(
-    term: SSHTerminal,
+    term: RemoteTerminal,
     console_callback: Callable[[], Console],
     app_config: AppConfig,
     display: str | None,

gambaterm/ssh_app_session.py

@@ -1,83 +1,21 @@
 """
-Provide an async context manager to create a blessed SSHTerminal
+Provide an async context manager to create a blessed RemoteTerminal
 from an AsyncSSH process.
 """
 from __future__ import annotations
 
 import os
-import codecs
 import asyncio
-import contextlib
 import subprocess
 from concurrent.futures import ThreadPoolExecutor
 from contextlib import asynccontextmanager, contextmanager
-from typing import AsyncIterator, Generator, IO, Iterator, TypeVar, Callable
-
-from blessed import Terminal as BlessedTerminal
-from blessed.terminal import WINSZ
+from typing import AsyncIterator, Iterator, TypeVar, Callable
 
 from asyncssh import SSHServerProcess
 
-T = TypeVar("T")
-
-
-# Python's curses.setupterm() can only be called once per process — subsequent
-# calls with a different terminal type are silently ignored. Since the SSH
-# server handles multiple concurrent connections in threads, all SSHTerminal
-# instances share whatever terminal type was initialized first by the local
-# Terminal(). We hardcode 'xterm-256color' as the kind since:
-#   1. It's universally compatible with modern terminals
-#   2. We use standard VT100/ANSI escape codes directly, not terminfo caps
-#   3. It avoids issues where the first SSH client's TERM value differs
-SSH_TERMINAL_TYPE = "xterm-256color"
-
-
-class SSHTerminal(BlessedTerminal):
-    """A blessed Terminal subclass for SSH streams.
-
-    Following the pattern from x84 (x84/terminal.py), this stubs raw/cbreak
-    mode (SSH is already raw) and overrides size detection to use values
-    provided by the SSH server.
-    """
+from .remote_terminal import RemoteTerminal
 
-    def __init__(
-        self,
-        stream: IO[str],
-        keyboard_fd: int,
-        rows: int,
-        columns: int,
-    ) -> None:
-        self._rows = rows
-        self._columns = columns
-        super().__init__(kind=SSH_TERMINAL_TYPE, stream=stream, force_styling=True)
-        # Blessed only sets _keyboard_fd when stream is sys.__stdout__, so
-        # for SSH pipes we must set it and initialize the decoder manually
-        self._keyboard_fd = keyboard_fd  # type: ignore[assignment]
-        self._keyboard_decoder = codecs.getincrementaldecoder("UTF-8")()
-
-    @property
-    def is_a_tty(self) -> bool:
-        return True
-
-    @contextlib.contextmanager
-    def raw(self) -> Generator[None, None, None]:
-        yield
-
-    @contextlib.contextmanager
-    def cbreak(self) -> Generator[None, None, None]:
-        yield
-
-    def _height_and_width(self) -> WINSZ:
-        return WINSZ(
-            ws_row=self._rows,
-            ws_col=self._columns,
-            ws_xpixel=0,
-            ws_ypixel=0,
-        )
-
-    def update_size(self, rows: int, columns: int) -> None:
-        self._rows = rows
-        self._columns = columns
+T = TypeVar("T")
 
 
 @asynccontextmanager
@@ -131,7 +69,7 @@ async def _input_pipe_from_process(
 
 @contextmanager
 def _bind_resize(
-    process: SSHServerProcess[str], ssh_term: SSHTerminal
+    process: SSHServerProcess[str], ssh_term: RemoteTerminal
 ) -> Iterator[None]:
     original_method = process.terminal_size_changed
 
@@ -151,9 +89,9 @@ def terminal_size_changed(
 async def process_to_terminal(
     process: SSHServerProcess[str],
     executor: ThreadPoolExecutor,
-    target: Callable[[SSHTerminal], T],
+    target: Callable[[RemoteTerminal], T],
 ) -> T:
-    """Create a blessed SSHTerminal from an SSH process.
+    """Create a blessed RemoteTerminal from an SSH process.
 
     Once the redirections are set up, I/O become synchronous,
     so we run the target function in a thread executor to avoid blocking the event loop
@@ -164,7 +102,7 @@ async def process_to_terminal(
 
     def _target() -> T:
         with open(write_fd, "w", newline="\r\n") as stream:
-            ssh_term = SSHTerminal(
+            ssh_term = RemoteTerminal(
                 stream=stream,
                 keyboard_fd=keyboard_fd,
                 rows=height,

gambaterm/telnet.py

@@ -23,8 +23,8 @@
 from .console import Console, InputGetter, GameboyColor
 from .keyboard_input import MESSAGE_SUGGESTING_KITTY_SUPPORT
 from .telnet_input import TelnetInputState, read_telnet_input
+from .remote_terminal import RemoteTerminal
 from .telnet_app_session import (
-    TelnetTerminal,
     set_tcp_nodelay,
     telnet_to_terminal,
 )
@@ -62,13 +62,13 @@ def _save_dir_name(username: str | None) -> str:
 
 
 def thread_target(
-    term: TelnetTerminal,
+    term: RemoteTerminal,
     console_callback: Callable[[], Console],
     app_config: AppConfig,
     color_mode: ColorMode,
     input_state: TelnetInputState | None = None,
 ) -> int:
-    """Run the emulator in a thread with the given TelnetTerminal."""
+    """Run the emulator in a thread with the given RemoteTerminal."""
     console: Console = console_callback()
 
     if app_config.input_file is not None:
@@ -345,7 +345,7 @@ async def _telnet_shell(
         console_callback = console_cls.pop_console_arguments(namespace)
         config = AppConfig(**vars(namespace))
 
-        def target(term: TelnetTerminal) -> int:
+        def target(term: RemoteTerminal) -> int:
             return thread_target(term, console_callback, config, color_mode, state)
 
         return await telnet_to_terminal(

gambaterm/telnet_app_session.py

@@ -1,74 +1,21 @@
 """
-Provide a blessed TelnetTerminal and paced output forwarding
-for the telnet server.
+Provide paced output forwarding and terminal setup for the telnet server.
 """
-
 from __future__ import annotations
 
 import os
 import socket
-import codecs
 import asyncio
-import contextlib
 from contextlib import contextmanager
-from typing import IO, TYPE_CHECKING, Callable, Generator, Iterator, TypeVar
+from typing import TYPE_CHECKING, Callable, Iterator, TypeVar
 from concurrent.futures import ThreadPoolExecutor
 
-from blessed import Terminal as BlessedTerminal
-from blessed.terminal import WINSZ
-
 if TYPE_CHECKING:
     from telnetlib3.stream_writer import TelnetWriter
 
-T = TypeVar("T")
-
-# See ssh_app_session.py for rationale on hardcoding the terminal type.
-TELNET_TERMINAL_TYPE = "xterm-256color"
-
-
-class TelnetTerminal(BlessedTerminal):
-    """A blessed Terminal subclass for telnet streams.
-
-    Stubs raw/cbreak mode (telnet is already raw) and overrides size
-    detection to use values from NAWS negotiation.
-    """
-
-    def __init__(
-        self,
-        stream: IO[str],
-        keyboard_fd: int,
-        rows: int,
-        columns: int,
-    ) -> None:
-        self._rows = rows
-        self._columns = columns
-        super().__init__(kind=TELNET_TERMINAL_TYPE, stream=stream, force_styling=True)
-        self._keyboard_fd = keyboard_fd  # type: ignore[assignment]
-        self._keyboard_decoder = codecs.getincrementaldecoder("UTF-8")()
-
-    @property
-    def is_a_tty(self) -> bool:
-        return True
-
-    @contextlib.contextmanager
-    def raw(self) -> Generator[None, None, None]:
-        yield
+from .remote_terminal import RemoteTerminal
 
-    @contextlib.contextmanager
-    def cbreak(self) -> Generator[None, None, None]:
-        yield
-
-    def _height_and_width(self) -> WINSZ:
-        return WINSZ(
-            ws_row=self._rows,
-            ws_col=self._columns,
-            ws_xpixel=0,
-            ws_ypixel=0,
-        )
-
-    def update_size(self, rows: int, columns: int) -> None:
-        self._rows = rows
-        self._columns = columns
+T = TypeVar("T")
 
 
 def set_tcp_nodelay(writer: TelnetWriter) -> None:
@@ -136,9 +83,9 @@ async def paced_forward_output(
 @contextmanager
 def bind_resize_telnet(
     writer: TelnetWriter,
-    term: TelnetTerminal,
+    term: RemoteTerminal,
 ) -> Iterator[None]:
-    """Hook telnetlib3 NAWS callbacks to update TelnetTerminal size.
+    """Hook telnetlib3 NAWS callbacks to update RemoteTerminal size.
 
     telnetlib3 dispatches NAWS via a callback registry on the writer
     (``_ext_callback[NAWS]``), not through ``protocol.on_naws`` directly.
@@ -162,16 +109,16 @@ def on_naws(rows: int, cols: int) -> None:
 async def telnet_to_terminal(
     writer: TelnetWriter,
     executor: ThreadPoolExecutor,
-    target: Callable[[TelnetTerminal], T],
+    target: Callable[[RemoteTerminal], T],
     input_read_fd: int,
 ) -> T:
-    """Create a TelnetTerminal and run *target* in a thread executor.
+    """Create a RemoteTerminal and run *target* in a thread executor.
 
     Sets up a pipe for output forwarding with paced delivery at 60 pps.
 
     :param writer: telnetlib3 writer
     :param executor: ThreadPoolExecutor for running the game thread
-    :param target: callable receiving the TelnetTerminal, run in executor
+    :param target: callable receiving the RemoteTerminal, run in executor
     :param input_read_fd: read end of the input pipe (keyboard_fd for blessed)
     :returns: return value of *target*
     """
@@ -183,7 +130,7 @@ async def telnet_to_terminal(
 
     def _target() -> T:
         with open(write_fd, "w", newline="\r\n") as stream:
-            telnet_term = TelnetTerminal(
+            telnet_term = RemoteTerminal(
                 stream=stream,
                 keyboard_fd=input_read_fd,
                 rows=rows,