pacai.ui.text

  1import atexit
  2import queue
  3import sys
  4import termios  # pylint: disable=import-error
  5import threading
  6import typing
  7
  8import pacai.core.action
  9import pacai.core.board
 10import pacai.core.gamestate
 11import pacai.core.ui
 12
 13class TextStreamUserInputDevice(pacai.core.ui.UserInputDevice):
 14    """
 15    A user input device that watches a text stream for input.
 16    The text stream could be a wide range of things,
 17    including a file or stdin.
 18    The target text stream will be closed when this device is closed.
 19    If the target stream is a tty (e.g., stdin),
 20    then the tty's attributes will be adjusted for streaming input.
 21    """
 22
 23    def __init__(self,
 24            input_stream: typing.TextIO,
 25            char_mapping: dict[str, pacai.core.action.Action] | None = None,
 26            **kwargs: typing.Any) -> None:
 27        self._input_stream: typing.TextIO = input_stream
 28        """ Where to get input from. """
 29
 30        if (char_mapping is None):
 31            char_mapping = pacai.core.ui.DUAL_CHAR_MAPPING
 32
 33        self._char_mapping: dict[str, pacai.core.action.Action] = char_mapping
 34        """ Map characters to actions. """
 35
 36        self._chars_queue: queue.Queue = queue.Queue()
 37        """ Used to store characters coming from the input stream. """
 38
 39        self._thread: threading.Thread = threading.Thread(target = _watch_text_stream, args = (self._input_stream, self._chars_queue))
 40        """ The thread that does the actual reading. """
 41
 42        self._old_settings: list | None = None
 43        """ Keep track of the old tty settings so we can reset properly. """
 44
 45        self._set_tty_attributes()
 46
 47        self._thread.start()
 48
 49    def get_inputs(self) -> list[pacai.core.action.Action]:
 50        output: list[pacai.core.action.Action] = []
 51        while (not self._chars_queue.empty()):
 52            char = self._chars_queue.get(block = False)
 53            if (char in self._char_mapping):
 54                output.append(self._char_mapping[char])
 55
 56        return output
 57
 58    def close(self) -> None:
 59        self._reset_tty_attributes()
 60        self._input_stream.close()
 61        self._thread.join()
 62
 63        super().close()
 64
 65    def __dict__(self) -> dict[str, typing.Any]:  # type: ignore[override]
 66        """ Do not allow for serialization because of the threads and streams. """
 67
 68        raise ValueError(f"This class ('{type(self).__qualname__}') cannot be serialized.")
 69
 70    def _set_tty_attributes(self) -> None:
 71        """ If the target stream is a tty, then set properties for better streaming input. """
 72
 73        # If we are not a tty, then there is nothing special to do.
 74        if (not self._input_stream.isatty()):
 75            return
 76
 77        # Do a platform check for POSIX.
 78        if (sys.platform.startswith("win")):
 79            raise ValueError("Terminal (tty) user input devices are not supported on Windows.")
 80
 81        self._old_settings = termios.tcgetattr(self._input_stream)
 82
 83        # Since the behavior of the terminal can be changed by this class,
 84        # ensure everything is reset when the program exits.
 85        atexit.register(self._reset_tty_attributes)
 86
 87        new_settings = termios.tcgetattr(self._input_stream)
 88
 89        # Modify lflags.
 90        # Remove echo and canonical (line-by-line) mode.
 91        new_settings[3] = new_settings[3] & ~(termios.ECHO | termios.ICANON)
 92
 93        # Modify CC flags.
 94        # Set non-canonical mode min chars and timeout.
 95        new_settings[6][termios.VMIN] = 1
 96        new_settings[6][termios.VTIME] = 0
 97
 98        termios.tcsetattr(self._input_stream, termios.TCSAFLUSH, new_settings)
 99
100    def _reset_tty_attributes(self) -> None:
101        if (self._old_settings is not None):
102            # Note that Windows does not have these settings.
103            if (hasattr(termios, 'tcsetattr') and hasattr(termios, 'TCSADRAIN')):
104                termios.tcsetattr(self._input_stream, termios.TCSADRAIN, self._old_settings)  # type: ignore[attr-defined,unused-ignore]
105
106            self._old_settings = None
107
108def _watch_text_stream(input_stream: typing.TextIO, result_queue: queue.Queue) -> None:
109    """ A thread worker to watch a text stream and relay the input. """
110
111    while (True):
112        try:
113            next_char = input_stream.read(1)
114        except ValueError:
115            # Should indicate that the stream was closed.
116            return
117
118        # Check for an EOF.
119        if ((next_char is None) or (len(next_char) == 0)):
120            return
121
122        result_queue.put(next_char, block = False)
123
124class TextUI(pacai.core.ui.UI):
125    """
126    A simple UI that renders the game to a text stream and takes input from another text stream.
127    This UI will be simple and generally meant for debugging.
128    """
129
130    def __init__(self,
131            input_stream: typing.TextIO,
132            output_stream: typing.TextIO,
133            **kwargs: typing.Any) -> None:
134        input_device = TextStreamUserInputDevice(input_stream, **kwargs)
135        super().__init__(user_input_device = input_device, **kwargs)
136
137        self._output_stream: typing.TextIO = output_stream
138        """ The stream output will be sent to. """
139
140    def draw(self, state: pacai.core.gamestate.GameState, **kwargs: typing.Any) -> None:
141        grid = state.board.to_grid()
142        for row in grid:
143            line = ''.join([self._translate_marker(marker, state) for marker in row])
144            self._output_stream.write(line + "\n")
145
146        self._output_stream.write(f"Score: {state.score}\n")
147
148        if (state.game_over):
149            self._output_stream.write('Game Over!\n')
150
151        self._output_stream.write("\n")
152        self._output_stream.flush()
153
154    def _translate_marker(self, marker: pacai.core.board.Marker, state: pacai.core.gamestate.GameState) -> str:
155        """
156        Convert a marker to a string.
157        This is generally trivial (since a marker is already a string),
158        but this allows children to implement special conversions.
159        """
160
161        return marker
162
163class StdioUI(TextUI):
164    """
165    A convenience class for a TextUI using stdin and stdout.
166    """
167
168    def __init__(self, **kwargs: typing.Any) -> None:
169        super().__init__(sys.stdin, sys.stdout, **kwargs)