refactor away getattr()'s

Author: jquast

gambaterm/blessed_keyboard_input.py

@@ -2,6 +2,7 @@
 
 import re
 from contextlib import contextmanager
+from dataclasses import dataclass, field
 from typing import Callable, Iterator
 
 from blessed import Terminal
@@ -12,6 +13,15 @@
 
 _CPR_RE = re.compile(r"\x1b\[\d+;\d+R")
 
+
+@dataclass
+class KeyboardState:
+    """Snapshot returned by a keyboard polling function."""
+
+    pressed: set[DomCode] = field(default_factory=set)
+    cpr_received: bool = False
+
+
 # Blessed synthesizes key_name as "KEY_{char}" for A-Z and 0-9 on release
 # and repeat events, so we need mappings for both functional keys and
 # printable keys that are used in game input/event mappings.
@@ -50,18 +60,18 @@ def keystroke_to_dom_code(keystroke: Keystroke) -> DomCode | None:
 @contextmanager
 def blessed_key_pressed_context(
     term: Terminal,
-) -> Iterator[Callable[[], set[DomCode]]]:
+) -> Iterator[Callable[[], KeyboardState]]:
     """Context manager providing a get_pressed() callable using blessed's kitty protocol."""
-    pressed: set[DomCode] = set()
+    state = KeyboardState()
 
     with term.enable_kitty_keyboard(
         report_events=True,
         report_alternates=True,
         report_all_keys=True,
     ):
 
-        def get_pressed() -> set[DomCode]:
-            get_pressed.cpr_received = False
+        def get_pressed() -> KeyboardState:
+            state.cpr_received = False
             while True:
                 key = term.inkey(timeout=0)
                 if not key:
@@ -79,20 +89,18 @@ def get_pressed() -> set[DomCode]:
                     raise OSError
                 # Cursor position response
                 if _CPR_RE.match(str(key)):
-                    get_pressed.cpr_received = True
+                    state.cpr_received = True
                     continue
                 dom_code = keystroke_to_dom_code(key)
                 if dom_code is None:
                     continue
                 if key.released:
-                    pressed.discard(dom_code)
+                    state.pressed.discard(dom_code)
                 else:
-                    pressed.add(dom_code)
-            return pressed
-
-        get_pressed.cpr_received = False
+                    state.pressed.add(dom_code)
+            return state
 
         try:
             yield get_pressed
         finally:
-            pressed.clear()
+            state.pressed.clear()

gambaterm/keyboard_input.py

@@ -10,7 +10,7 @@
 
 from .dom_codes import DomCode
 from .console import Console, InputGetter
-from .blessed_keyboard_input import blessed_key_pressed_context
+from .blessed_keyboard_input import KeyboardState, blessed_key_pressed_context
 from .pynput_keyboard_input import pynput_key_pressed_context
 from .x11_keyboard_input import x11_key_pressed_context
 
@@ -43,6 +43,10 @@ def get_input_mapping(console: Console) -> dict[DomCode, Console.Input]:
         DomCode.ARROW_DOWN: console.Input.DOWN,
         DomCode.ARROW_LEFT: console.Input.LEFT,
         DomCode.ARROW_RIGHT: console.Input.RIGHT,
+        DomCode.NUMPAD8: console.Input.UP,
+        DomCode.NUMPAD2: console.Input.DOWN,
+        DomCode.NUMPAD4: console.Input.LEFT,
+        DomCode.NUMPAD6: console.Input.RIGHT,
         DomCode.US_Z: console.Input.A,
         DomCode.US_X: console.Input.B,
         # WASD controls
@@ -72,31 +76,46 @@ def get_event_mapping(console: Console) -> dict[DomCode, Console.Event]:
     }
 
 
-def make_get_input(
-    console: Console,
-    get_pressed: Callable[[], set[DomCode]],
-) -> InputGetter:
-    current_pressed: set[DomCode] = set()
-    input_mapping = get_input_mapping(console)
-    event_mapping = get_event_mapping(console)
-
-    def get_input() -> set[Console.Input]:
-        nonlocal current_pressed
-        old_pressed, current_pressed = current_pressed, set(get_pressed())
-        # Propagate CPR flag from keyboard handler to run loop
-        get_input.cpr_received = getattr(get_pressed, "cpr_received", False)
-        for event in map(event_mapping.get, current_pressed - old_pressed):
+class GameInputGetter:
+    """Callable that translates raw key state into console inputs."""
+
+    def __init__(
+        self,
+        console: Console,
+        get_pressed: Callable[[], set[DomCode] | KeyboardState],
+    ) -> None:
+        self._get_pressed = get_pressed
+        self._current_pressed: set[DomCode] = set()
+        self._input_mapping = get_input_mapping(console)
+        self._event_mapping = get_event_mapping(console)
+        self._console = console
+        self.cpr_state = KeyboardState()
+
+    def __call__(self) -> set[Console.Input]:
+        result = self._get_pressed()
+        if isinstance(result, KeyboardState):
+            self.cpr_state.cpr_received = result.cpr_received
+            new_pressed = set(result.pressed)
+        else:
+            self.cpr_state.cpr_received = False
+            new_pressed = set(result)
+        old_pressed, self._current_pressed = self._current_pressed, new_pressed
+        for event in map(self._event_mapping.get, new_pressed - old_pressed):
             if event is None:
                 continue
-            console.handle_event(event)
+            self._console.handle_event(event)
         return {
-            input_mapping[keysym]
-            for keysym in current_pressed
-            if keysym in input_mapping
+            self._input_mapping[keysym]
+            for keysym in self._current_pressed
+            if keysym in self._input_mapping
         }
 
-    get_input.cpr_received = False
-    return get_input
+
+def make_get_input(
+    console: Console,
+    get_pressed: Callable[[], set[DomCode] | KeyboardState],
+) -> GameInputGetter:
+    return GameInputGetter(console, get_pressed)
 
 
 def _kitty_supported(term: Terminal) -> bool:
@@ -153,7 +172,7 @@ def console_input_from_keyboard_context(
         if xdg_session_type is None:
             xdg_session_type = os.environ.get("XDG_SESSION_TYPE", "")
         if xdg_session_type != "x11":
-            raise RuntimeError(MESSAGE_FOR_WAYLAND_USERS)
+            raise RuntimeError(MESSAGE_SUGGESTING_KITTY_SUPPORT)
         with console_input_from_x11_keyboard_context(console, display) as get_input:
             yield get_input
     else:
@@ -166,7 +185,7 @@ def key_pressed_context(
     term: Terminal,
     display: str | None = None,
     xdg_session_type: str | None = None,
-) -> Iterator[Callable[[], set[DomCode]]]:
+) -> Iterator[Callable[[], set[DomCode] | KeyboardState]]:
     if _kitty_supported(term):
         with blessed_key_pressed_context(term) as get_pressed:
             yield get_pressed
@@ -193,7 +212,11 @@ def main() -> None:
             with key_pressed_context(term) as get_pressed:
                 while True:
                     # Get codes
-                    codes = (x.value for x in get_pressed())
+                    result = get_pressed()
+                    pressed = result.pressed if isinstance(
+                        result, KeyboardState
+                    ) else result
+                    codes = (x.value for x in pressed)
                     # Print pressed key codes
                     print("\r", *codes, flush=True, end=term.clear_eol)
                     # Tick

gambaterm/main.py

@@ -140,10 +140,6 @@ def main(
             f"Invalid color mode `{args.color_mode}`: the value must be between 1 and 4"
         )
 
-    # Query terminal background color BEFORE entering kitty keyboard mode,
-    # as the OSC 11 response would leak into the keyboard input stream.
-    bg_color = term.get_bgcolor(bits=8)
-
     # Enter terminal raw mode
     with term.raw():
         try:
@@ -178,7 +174,6 @@ def main(
                         break_after=args.break_after,
                         speed_factor=args.speed_factor,
                         use_cpr_sync=args.cpr_sync,
-                        bg_color=bg_color,
                     )
 
         # Deal with ctrl+c and ctrl+d exceptions

gambaterm/pynput_keyboard_input.py

@@ -34,8 +34,6 @@ def get_value_from_pynput_key_code(key: pynput.keyboard.KeyCode) -> DomCode | No
         assert False
 
 
-
-
 @contextmanager
 def pynput_key_pressed_context() -> Iterator[Callable[[], set[DomCode]]]:
     import pynput.keyboard

gambaterm/run.py

@@ -14,6 +14,7 @@
 from .termblit import blit
 from .audio import AudioOut
 from .console import Console, InputGetter
+from .keyboard_input import GameInputGetter
 from .colors import ColorMode
 
 
@@ -53,19 +54,11 @@ def run(
     break_after: int | None = None,
     speed_factor: float = 1.0,
     use_cpr_sync: bool = False,
-    bg_color: tuple[int, int, int] = (-1, -1, -1),
 ) -> None:
     assert color_mode > 0
 
     # Prepare buffers with invalid data
-    # Use terminal background color so the blitter can skip pixels that
-    # already match it on the first frame
-    r, g, b = bg_color
-    if r == -1:
-        bg_val = np.uint32(0)  # assume black if detection fails
-    else:
-        bg_val = np.uint32((r << 16) | (g << 8) | b)
-    video = np.full((console.HEIGHT, console.WIDTH), bg_val, np.uint32)
+    video = np.full((console.HEIGHT, console.WIDTH), 0, np.uint32)
     audio = np.full((2 * console.TICKS_IN_FRAME, 2), -0x7FFF, np.int16)
     last_frame = video.copy()
 
@@ -92,6 +85,13 @@ def run(
     if audio_out:
         start -= 0.1
 
+    # Resolve CPR state once (only GameInputGetter has it)
+    cpr_state = (
+        get_input.cpr_state
+        if isinstance(get_input, GameInputGetter)
+        else None
+    )
+
     # Prepare state
     new_frame = False
     screen_ready = True
@@ -121,7 +121,7 @@ def run(
                 audio_out.send(audio[:samples, :])
 
         # Check for CPR response (set by keyboard handler during get_input)
-        if use_cpr_sync and getattr(get_input, "cpr_received", False):
+        if use_cpr_sync and cpr_state is not None and cpr_state.cpr_received:
             screen_ready = True
 
         # Render video
@@ -133,20 +133,22 @@ def run(
                 # Check terminal size
                 new_height = term.height or 24
                 new_width = term.width or 80
-                maybe_clear_prefix, maybe_clear_suffix = b"", b""
+                maybe_clear_seq = b""
                 if (new_height, new_width) != (height, width):
-                    # Write video frame with clear sequence inside synchronized output mode (DEC
-                    # 2026) to prevent flicker, and, set last_frame as inverse of the current frame
-                    # to ensure a full redraw by the blitter
-                    maybe_clear_prefix = b"\033[?2026h\033[H\033[2J"
-                    maybe_clear_suffix = b"\033[?2026l"
+                    maybe_clear_seq = b"\033[H\033[2J"
                     height, width = new_height, new_width
                     refx, refy = get_ref(width, height, console)
                     last_frame = ~video
-                # Render frame
-                video_data = maybe_clear_prefix + blit(
-                    video, last_frame, refx, refy, width - 1, height, color_mode
-                ) + maybe_clear_suffix
+                # Render frame with synchronized output mode (DEC 2026) to prevent flickering
+                # when the screen is cleared, or an artificial CRT-like "rolling band" side-effects
+                # from fast "sprite blinking" meant to cause "transparency" effect on original HW,
+                # https://zladx.github.io/posts/links-awakening-partial-translucency
+                video_data = (
+                    b"\033[?2026h"
+                    + maybe_clear_seq
+                    + blit(video, last_frame, refx, refy, width - 1, height, color_mode)
+                    + b"\033[?2026l"
+                )
                 last_frame = video.copy()
                 # Update reporting
                 data_length.append(len(video_data))

gambaterm/ssh.py

@@ -111,7 +111,6 @@ async def ssh_process_handler(process: SSHServerProcess[str]) -> int:
     executor: ThreadPoolExecutor = process.get_extra_info("executor")
     display = process.channel.get_x11_display()
     command = process.channel.get_command()
-    environment = dict(process.channel.get_environment())
     terminal_type = process.get_terminal_type()
     connection = process.get_extra_info("connection")
     username = process.get_extra_info("username")
@@ -182,17 +181,6 @@ async def ssh_process_handler(process: SSHServerProcess[str]) -> int:
     else:
         color_mode = ColorMode.HAS_24_BIT_COLOR
 
-    if color_mode == ColorMode.NO_COLOR:
-        print(
-            f"Your terminal `{terminal_type}` doesn't seem to support colors.",
-            f"Try to force a color mode by appending `-t -- --color-mode 3`"
-            f" to the ssh command",
-            sep="\r\n",
-            file=process.stdout,
-        )
-        print(f"< User `{username}` terminal `{terminal_type}` does not support colors")
-        return 1
-
     # Now is a good time to instanciate the console
     # (it might fail if the ROM does not exist for instance)
     console = console_callback()