Add frontend callback (PR #46)

Also:
- add proper logging using structlog for the servers (#37)
- refactor configuration handling and keyboard detection for the servers.

Author: vxgmichel

gambaterm/console.py

@@ -4,7 +4,10 @@
 from pathlib import Path
 import tempfile
 from enum import IntEnum
-from typing import Callable
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from .main import AppConfig
 
 import numpy as np
 import numpy.typing as npt
@@ -53,11 +56,8 @@ def add_console_arguments(cls, parser: argparse.ArgumentParser) -> None:
         pass
 
     @classmethod
-    def pop_console_arguments(
-        cls, namespace: argparse.Namespace
-    ) -> Callable[[], Console]:
-        romfile: Path = namespace.romfile
-        return lambda: cls(romfile)
+    def from_app_config(cls, app_config: AppConfig) -> Console:
+        return cls(app_config.romfile)
 
     def __init__(self, romfile: Path):
         self.romfile = str(romfile.resolve())
@@ -117,17 +117,17 @@ def add_console_arguments(cls, parser: argparse.ArgumentParser) -> None:
         )
 
     @classmethod
-    def pop_console_arguments(
-        cls, namespace: argparse.Namespace
-    ) -> Callable[[], Console]:
-        romfile: Path = namespace.romfile
-        input_file: Path | None = namespace.input_file
-        save_directory: Path | None = namespace.save_directory
-        force_gameboy: bool = namespace.__dict__.pop("force_gameboy")
+    def from_app_config(cls, app_config: AppConfig) -> GameboyColor:
+        romfile: Path = app_config.romfile
+        input_file: Path | None = app_config.input_file
+        save_directory: Path | None = app_config.save_directory
+        force_gameboy: bool = getattr(
+            app_config.console_namespace, "force_gameboy", False
+        )
         # Save directory defaults to the rom file directory (unless we read the input from a file)
         if input_file is None and save_directory is None:
             save_directory = romfile.parent
-        return lambda: cls(romfile, save_directory, force_gameboy)
+        return cls(romfile, save_directory, force_gameboy)
 
     def __init__(
         self,

gambaterm/main.py

@@ -4,8 +4,9 @@
 import time
 import argparse
 from pathlib import Path
-from typing import ContextManager
-from dataclasses import dataclass
+from typing import ContextManager, TYPE_CHECKING
+import dataclasses
+from dataclasses import dataclass, field
 
 from blessed import Terminal
 
@@ -18,6 +19,10 @@
 from .controller_input import combine_console_input_from_controller_context
 from .file_input import console_input_from_file_context, write_input_context
 
+# `typing.Self` is not available in python 3.10
+if TYPE_CHECKING:
+    from typing import Self
+
 
 @dataclass
 class AppConfig:
@@ -29,13 +34,26 @@ class AppConfig:
     speed: float
     skip_inputs: int
     cpr_sync: bool
-    save_directory: Path | None
+    save_directory: Path | None = None
+    console_namespace: argparse.Namespace = field(default_factory=argparse.Namespace)
+
+    @classmethod
+    def from_namespace(cls, namespace: argparse.Namespace) -> Self:
+        allowed_keys = {
+            f.name for f in dataclasses.fields(cls) if f.name != "console_namespace"
+        }
+        kwargs = {k: v for k, v in vars(namespace).items() if k in allowed_keys}
+        console_keys = {
+            k: v for k, v in vars(namespace).items() if k not in allowed_keys
+        }
+        kwargs["console_namespace"] = argparse.Namespace(**console_keys)
+        return cls(**kwargs)
 
 
 @dataclass
 class LocalAppConfig(AppConfig):
-    enable_controller: bool
-    write_input: Path | None
+    enable_controller: bool = False
+    write_input: Path | None = None
 
 
 def add_base_arguments(parser: argparse.ArgumentParser) -> None:
@@ -138,16 +156,15 @@ def main(
 
     # Parse arguments
     namespace = parser.parse_args(parser_args)
-    disable_audio: bool = namespace.__dict__.pop("disable_audio")
-    console_callback = console_cls.pop_console_arguments(namespace)
-    args = LocalAppConfig(**vars(namespace))
+    disable_audio = getattr(namespace, "disable_audio", False)
+    args = LocalAppConfig.from_namespace(namespace)
 
     # Check that the ROM file exists
     if not args.romfile.exists():
         raise SystemExit(f"ROM file `{args.romfile}` does not exist")
 
     # Instantiate the console and terminal
-    console = console_callback()
+    console = console_cls.from_app_config(args)
     terminal = Terminal()
 
     # Prepare input context

gambaterm/remote_terminal.py

@@ -1,17 +1,22 @@
 """
-Provide a blessed Terminal subclass for remote (SSH/telnet) streams.
+Provide common resources for both Telnet and SSH terminals.
 """
 
 from __future__ import annotations
 
 import codecs
+from concurrent.futures import ThreadPoolExecutor, CancelledError
+from enum import Enum
 import hashlib
 import contextlib
-from typing import IO, Generator
+from typing import IO, Callable, Generator, TypeAlias, TYPE_CHECKING
 
 from blessed import Terminal as BlessedTerminal
 from blessed.terminal import WINSZ
 
+if TYPE_CHECKING:
+    from .main import AppConfig
+
 
 class RemoteTerminal(BlessedTerminal):
     """A blessed Terminal subclass for remote streams (SSH, telnet).
@@ -87,6 +92,55 @@ def update_size(self, rows: int, columns: int) -> None:
         self._columns = columns
 
 
+class KeyboardSupport(Enum):
+    BASIC = "basic"
+    KEYBOARD_PROTOCOL = "keyboard_protocol"
+    X11 = "x11"
+
+
+class KeyboardSupportDetection:
+    def __init__(
+        self,
+        terminal: RemoteTerminal,
+        display: str | None = None,
+        executor: ThreadPoolExecutor | None = None,
+    ) -> None:
+        self.terminal = terminal
+        self.display = display
+        self.executor = executor
+        self._cache: KeyboardSupport | None = None
+
+    def get(self, timeout: float = 3.0) -> KeyboardSupport:
+        if self._cache is not None:
+            return self._cache
+        self._cache = self._detect(timeout)
+        return self._cache
+
+    def _detect(self, timeout: float = 3.0) -> KeyboardSupport:
+        from .keyboard_input import is_kitty_keyboard_protocol_supported
+
+        if is_kitty_keyboard_protocol_supported(self.terminal, timeout=timeout):
+            return KeyboardSupport.KEYBOARD_PROTOCOL
+
+        elif self.display and self.executor:
+            from .x11_keyboard_input import is_x11_display_functional
+
+            try:
+                if self.executor.submit(is_x11_display_functional, self.display).result(
+                    timeout=timeout
+                ):
+                    return KeyboardSupport.X11
+            except CancelledError:
+                pass
+
+        return KeyboardSupport.BASIC
+
+
+FrontendCallback: TypeAlias = Callable[
+    [RemoteTerminal, "AppConfig", KeyboardSupportDetection], "AppConfig"
+]
+
+
 def user_directory_name(username: str | None) -> str:
     """Hash the username into a safe directory name.