Merge remote-tracking branch 'upstream/main' into jq/add-telnet-server

Author: jquast

README.md

@@ -219,8 +219,8 @@ The key bindings are not configurable at the moment:
 | Buttons    | Keyboard with arrows | Keyboard with WASD | Controller            |
 |------------|----------------------|--------------------|-----------------------|
 | Directions | Arrows               | W A S D            | Left hat / Left stick |
-| A          | Z                    | K                  | Button 0 / Button 3   |
-| B          | X                    | J                  | Button 1 / Button 2   |
+| A          | X                    | K                  | Button 0 / Button 3   |
+| B          | Z                    | J                  | Button 1 / Button 2   |
 | Start      | Enter                | Enter              | Button 7              |
 | Select     | Right Shift          | Right Shift        | Button 6              |
 

gambaterm/audio.py

@@ -77,10 +77,20 @@ def audio_player(
     console: Console, speed_factor: float = 1.0
 ) -> Iterator[AudioOut | None]:
     # Perform late imports
-    # Those can fail if a linux machine doesn't have portaudio or libsamplerate
-    # installed
     import samplerate
-    import sounddevice
+
+    # Especially for sounddevice, as it doesn't package the portaudio library in its manylinux wheels.
+    try:
+        import sounddevice
+    except OSError:
+        raise SystemExit(
+            """\
+Audio output is not available because the PortAudio library could not be found.
+Please make sure you have portaudio installed.
+For example, on Debian-based distributions, you can run:
+  $ sudo apt install libportaudio2
+Otherswise, you can use the --no-audio option to run without audio support."""
+        )
 
     input_rate = console.FPS * console.TICKS_IN_FRAME
     resampler = samplerate.Resampler("linear", channels=2)

gambaterm/blessed_keyboard_input.py

@@ -1,8 +1,6 @@
 from __future__ import annotations
 
-import re
 from contextlib import contextmanager
-from dataclasses import dataclass, field
 from typing import Callable, Iterator
 
 from blessed import Terminal
@@ -11,17 +9,6 @@
 from .dom_codes import DomCode
 from .keys import ASCII_PRINTABLE_TO_DOM_CODE
 
-_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
-    keystrokes: "list[Keystroke]" = field(default_factory=list)
-
 
 # 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
@@ -61,38 +48,42 @@ def keystroke_to_dom_code(keystroke: Keystroke) -> DomCode | None:
 @contextmanager
 def blessed_key_pressed_context(
     term: Terminal,
-) -> Iterator[Callable[[], KeyboardState]]:
+) -> Iterator[tuple[Callable[[], set[DomCode]], Callable[[], list[Keystroke]]]]:
     """Context manager providing a get_pressed() callable using blessed's kitty protocol."""
-    state = KeyboardState()
+    pressed: set[DomCode] = set()
+    keystrokes: list[Keystroke] = []
 
     with term.enable_kitty_keyboard(
         report_events=True,
         report_alternates=True,
         report_all_keys=True,
     ):
 
-        def get_pressed() -> KeyboardState:
-            state.cpr_received = False
-            state.keystrokes.clear()
+        def _update() -> None:
             while True:
                 key = term.inkey(timeout=0)
                 if not key:
                     break
-                state.keystrokes.append(key)
-                # Cursor position response
-                if _CPR_RE.match(str(key)):
-                    state.cpr_received = True
-                    continue
+                keystrokes.append(key)
                 dom_code = keystroke_to_dom_code(key)
                 if dom_code is None:
                     continue
                 if key.released:
-                    state.pressed.discard(dom_code)
+                    pressed.discard(dom_code)
                 else:
-                    state.pressed.add(dom_code)
-            return state
+                    pressed.add(dom_code)
+
+        def get_pressed() -> set[DomCode]:
+            _update()
+            return pressed.copy()
+
+        def pop_keystrokes() -> list[Keystroke]:
+            nonlocal keystrokes
+            _update()
+            result, keystrokes = keystrokes, []
+            return result
 
         try:
-            yield get_pressed
+            yield get_pressed, pop_keystrokes
         finally:
-            state.pressed.clear()
+            pressed.clear()

gambaterm/colors.py

@@ -1,27 +1,13 @@
 from __future__ import annotations
 
-import os
 import sys
 from enum import IntEnum
 
 from blessed import Terminal
 
-# Terminals that support at least 16 colors but may not be detected
-# correctly by curses/terminfo.
-BASIC_TERMINALS = [
-    "screen",
-    "vt100",
-    "vt220",
-    "rxvt",
-    "color",
-    "ansi",
-    "cygwin",
-    "linux",
-]
-
 
 class ColorMode(IntEnum):
-    NO_COLOR = 0
+    COULD_NOT_DETECT = 0
     HAS_2_BIT_COLOR = 1
     HAS_4_BIT_COLOR = 2
     HAS_8_BIT_COLOR = 3
@@ -54,11 +40,7 @@ def detect_local_color_mode(term: Terminal) -> ColorMode:
         return ColorMode.HAS_4_BIT_COLOR
     if n >= 4:
         return ColorMode.HAS_2_BIT_COLOR
-    # Fallback for terminals that curses/terminfo under-reports
-    term_env = os.environ.get("TERM", "").lower()
-    if any(x in term_env for x in BASIC_TERMINALS):
-        return ColorMode.HAS_4_BIT_COLOR
-    return ColorMode.NO_COLOR
+    return ColorMode.COULD_NOT_DETECT
 
 
 def main() -> None:

gambaterm/console.py

@@ -4,7 +4,7 @@
 from pathlib import Path
 import tempfile
 from enum import IntEnum
-from typing import Callable, Set
+from typing import Callable
 
 import numpy as np
 import numpy.typing as npt
@@ -96,10 +96,6 @@ def handle_event(self, event: Event) -> None:
             assert False
 
 
-# Type Alias
-InputGetter = Callable[[], Set[Console.Input]]
-
-
 class GameboyColor(Console):
     WIDTH: int = 160
     HEIGHT: int = 144

gambaterm/controller_input.py

@@ -4,7 +4,8 @@
 from typing import Callable, ContextManager, Iterator
 from contextlib import contextmanager
 
-from .console import Console, InputGetter
+from .console import Console
+from .input_getter import BaseInputGetter, StackedInputGetter
 
 
 def get_controller_input_mapping(console: Console) -> dict[str, Console.Input]:
@@ -97,7 +98,9 @@ def get_pressed() -> set[str]:
 
 
 @contextmanager
-def console_input_from_controller_context(console: Console) -> Iterator[InputGetter]:
+def console_input_from_controller_context(
+    console: Console,
+) -> Iterator[Callable[[], set[Console.Input]]]:
     input_mapping = get_controller_input_mapping(console)
     event_mapping = get_controller_event_mapping(console)
     current_pressed: set[str] = set()
@@ -119,10 +122,23 @@ def get_gb_input() -> set[Console.Input]:
         yield get_gb_input
 
 
+class ControllerInputGetter(StackedInputGetter):
+    def __init__(
+        self,
+        base_getter: BaseInputGetter,
+        extra_get_pressed: Callable[[], set[Console.Input]],
+    ) -> None:
+        super().__init__(base_getter)
+        self._extra_get_pressed = extra_get_pressed
+
+    def get_pressed(self) -> set[Console.Input]:
+        return super().get_pressed() | self._extra_get_pressed()
+
+
 @contextmanager
 def combine_console_input_from_controller_context(
-    console: Console, context: ContextManager[InputGetter]
-) -> Iterator[InputGetter]:
-    with context as getter1:
-        with console_input_from_controller_context(console) as getter2:
-            yield lambda: getter1() | getter2()
+    context: ContextManager[BaseInputGetter],
+) -> Iterator[ControllerInputGetter]:
+    with context as base_getter:
+        with console_input_from_controller_context(base_getter.console) as getter2:
+            yield ControllerInputGetter(base_getter, getter2)

gambaterm/file_input.py

@@ -6,7 +6,24 @@
 from zipfile import ZipFile, BadZipFile
 from typing import ContextManager, Iterator
 
-from .console import Console, InputGetter
+from blessed import Terminal
+
+from .console import Console
+from .input_getter import BaseInputGetter, StackedInputGetter
+
+
+class FileInputGetter(BaseInputGetter):
+    def __init__(
+        self,
+        console: Console,
+        terminal: Terminal,
+        input_generator: Iterator[set[Console.Input]],
+    ) -> None:
+        super().__init__(console, terminal)
+        self._generator = input_generator
+
+    def get_pressed(self) -> set[Console.Input]:
+        return next(self._generator)
 
 
 def get_inputs_ref(console: Console) -> list[Console.Input]:
@@ -45,8 +62,8 @@ def open_input_log_file(path: Path) -> Iterator[TextIOWrapper]:
 
 @contextmanager
 def console_input_from_file_context(
-    console: Console, path: Path, skip_first_frames: int = 188
-) -> Iterator[InputGetter]:
+    console: Console, terminal: Terminal, path: Path, skip_first_frames: int = 188
+) -> Iterator[FileInputGetter]:
     inputs_ref = get_inputs_ref(console)
     with open_input_log_file(path) as f:
 
@@ -62,19 +79,24 @@ def gen() -> Iterator[set[Console.Input]]:
                 yield set()
 
         input_generator = gen()
-        yield lambda: next(input_generator)
+        yield FileInputGetter(console, terminal, input_generator)
+
+
+class WriteInputGetter(StackedInputGetter):
+    def __init__(self, base_getter: BaseInputGetter, file: TextIOWrapper) -> None:
+        super().__init__(base_getter)
+        self._file = file
+
+    def get_pressed(self) -> set[Console.Input]:
+        value = super().get_pressed()
+        print(value_to_line(self.console, value), file=self._file)
+        return value
 
 
 @contextmanager
 def write_input_context(
-    console: Console, context: ContextManager[InputGetter], path: Path
-) -> Iterator[InputGetter]:
+    context: ContextManager[BaseInputGetter], path: Path
+) -> Iterator[WriteInputGetter]:
     with open(path, "w") as f:
-        with context as getter:
-
-            def new_getter() -> set[Console.Input]:
-                value = getter()
-                print(value_to_line(console, value), file=f)
-                return value
-
-            yield new_getter
+        with context as base_getter:
+            yield WriteInputGetter(base_getter, f)

gambaterm/input_getter.py

@@ -0,0 +1,78 @@
+from blessed.keyboard import Keystroke
+from blessed.terminal import Terminal
+
+from .console import Console
+
+
+def pop_keystrokes_from_terminal(terminal: Terminal) -> list[Keystroke]:
+    return list(iter(lambda: terminal.inkey(timeout=0), ""))
+
+
+class BaseInputGetter:
+    """Base class for input getters.
+
+    This class reports both:
+    - the currently pressed console buttons
+    - the keystrokes that occurred since the last call
+
+    The reason why those apparently separated responsibilities are combined in the same class
+    is that some input sources (e.g. keyboard with kitty protocol) provide both information through the same API.
+
+    It provides default implementation for pop_keystrokes() that reads from the terminal,
+    which is useful for input sources that do not mess with the terminal input.
+
+    More practically:
+    - KittyInputGetter: reports pressed console buttons and keystrokes from the terminal using the kitty keyboard protocol
+    - X11InputGetter: reports pressed console buttons from X11 events, and reports keystrokes from the terminal
+    - PynputInputGetter: reports pressed console buttons from OS key hooks, and reports keystrokes from the terminal
+    - FileInputGetter: reports pressed console buttons from a file, and reports keystrokes from the terminal
+    """
+
+    def __init__(self, console: Console, terminal: Terminal) -> None:
+        self._console = console
+        self._terminal = terminal
+
+    @property
+    def console(self) -> Console:
+        return self._console
+
+    @property
+    def terminal(self) -> Terminal:
+        return self._terminal
+
+    def get_pressed(self) -> set[Console.Input]:
+        """Get the currently pressed inputs."""
+        raise NotImplementedError
+
+    def pop_keystrokes(self) -> list[Keystroke]:
+        """Get the keystrokes that occurred since the last call."""
+        return pop_keystrokes_from_terminal(self.terminal)
+
+
+class StackedInputGetter(BaseInputGetter):
+    """
+    BaseInputGetter that combines the pressed inputs of a base getter with extra pressed inputs from another callable.
+
+    This is useful to combine multiple input sources, e.g. keyboard and controller.
+
+    More pratically:
+    - ControllerInputGetter: add console inputs from a controller on top of another input getter
+    - WriteInputGetter: write the console inputs from another input getter to a file, without modifying them
+    """
+
+    def __init__(self, base_getter: BaseInputGetter) -> None:
+        self.base_getter = base_getter
+
+    @property
+    def console(self) -> Console:
+        return self.base_getter.console
+
+    @property
+    def terminal(self) -> Terminal:
+        return self.base_getter.terminal
+
+    def get_pressed(self) -> set[Console.Input]:
+        return self.base_getter.get_pressed()
+
+    def pop_keystrokes(self) -> list[Keystroke]:
+        return self.base_getter.pop_keystrokes()