pacai.core.ui

  1import abc
  2import argparse
  3import os
  4import time
  5import typing
  6
  7import PIL.Image
  8import PIL.ImageDraw
  9import PIL.ImageFont
 10import edq.util.time
 11
 12import pacai.core.action
 13import pacai.core.board
 14import pacai.core.font
 15import pacai.core.gamestate
 16import pacai.core.spritesheet
 17import pacai.util.alias
 18import pacai.util.reflection
 19
 20DEFAULT_FPS: int = 15
 21
 22DEFAULT_ANIMATION_FPS: int = 15
 23DEFAULT_ANIMATION_SKIP_FRAMES: int = 1
 24MIN_ANIMATION_FPS: int = 1
 25DEFAULT_ANIMATION_OPTIMIZE: bool = False
 26
 27DEFAULT_SPRITE_SHEET: str = 'generic'
 28
 29ANIMATION_KEY: str = 'UI.draw_image'
 30
 31ANIMATION_EXTS: list[str] = ['.gif', '.webp']
 32""" The allowed extensions for animation files. """
 33
 34WASD_CHAR_MAPPING: dict[str, pacai.core.action.Action] = {
 35    'w': pacai.core.action.NORTH,
 36    'W': pacai.core.action.NORTH,
 37    'ArrowUp': pacai.core.action.NORTH,
 38
 39    'a': pacai.core.action.WEST,
 40    'A': pacai.core.action.WEST,
 41    'ArrowLeft': pacai.core.action.WEST,
 42
 43    's': pacai.core.action.SOUTH,
 44    'S': pacai.core.action.SOUTH,
 45    'ArrowDown': pacai.core.action.SOUTH,
 46
 47    'd': pacai.core.action.EAST,
 48    'D': pacai.core.action.EAST,
 49    'ArrowRight': pacai.core.action.EAST,
 50
 51    ' ': pacai.core.action.STOP,
 52    'space': pacai.core.action.STOP,
 53    'Space': pacai.core.action.STOP,
 54    'SPACE': pacai.core.action.STOP,
 55}
 56""" A character to action mapping using the common WASD scheme. """
 57
 58ARROW_CHAR_MAPPING: dict[str, pacai.core.action.Action] = {
 59    'Up': pacai.core.action.NORTH,
 60    'ArrowUp': pacai.core.action.NORTH,
 61
 62    'Left': pacai.core.action.WEST,
 63    'ArrowLeft': pacai.core.action.WEST,
 64
 65    'Down': pacai.core.action.SOUTH,
 66    'ArrowDown': pacai.core.action.SOUTH,
 67
 68    'Right': pacai.core.action.EAST,
 69    'ArrowRight': pacai.core.action.EAST,
 70
 71    ' ': pacai.core.action.STOP,
 72    'space': pacai.core.action.STOP,
 73    'Space': pacai.core.action.STOP,
 74    'SPACE': pacai.core.action.STOP,
 75}
 76""" A character to action mapping using the arrow keys. """
 77
 78DUAL_CHAR_MAPPING: dict[str, pacai.core.action.Action] = WASD_CHAR_MAPPING | ARROW_CHAR_MAPPING
 79""" A character to action mapping that uses both WASD_CHAR_MAPPING and ARROW_CHAR_MAPPING. """
 80
 81class UserInputDevice(abc.ABC):
 82    """
 83    This class provides a way for users to convey inputs through a UI.
 84    Not all UIs will support user input.
 85    """
 86
 87    @abc.abstractmethod
 88    def get_inputs(self) -> list[pacai.core.action.Action]:
 89        """
 90        Get any inputs that have occurred since the last call to this method.
 91        This method is responsible for not returning the same input instance in subsequent calls.
 92        The last input in the returned list should be the most recent input.
 93        """
 94
 95    def close(self) -> None:
 96        """ Close the user input channel and release all owned resources. """
 97
 98class UI(abc.ABC):
 99    """
100    UIs represent the basic way that a game interacts with the user,
101    by displaying the state of the game and taking input from the user (if applicable).
102    """
103
104    def __init__(self,
105            user_input_device: UserInputDevice | None = None,
106            fps: int = DEFAULT_FPS,
107            animation_path: str | None = None,
108            animation_optimize: bool = DEFAULT_ANIMATION_OPTIMIZE,
109            animation_fps: int = DEFAULT_ANIMATION_FPS,
110            animation_skip_frames: int = DEFAULT_ANIMATION_SKIP_FRAMES,
111            sprite_sheet_path: str = DEFAULT_SPRITE_SHEET,
112            font_path: str = pacai.core.font.DEFAULT_FONT_PATH,
113            **kwargs: typing.Any) -> None:
114        self._user_input_device: UserInputDevice | None = user_input_device
115        """ The device to use to get user input. """
116
117        self._fps: int = fps
118        """
119        The desired frames per second this game will be displayed at.
120        Zero or lower values will be ignored.
121        This is just a suggestion that the game will try an accommodate.
122        Not all UIs will observe fps.
123        """
124
125        self._last_fps_wait: edq.util.time.Timestamp | None = None
126        """
127        Keep track of the last time the UI waited to adjust the fps.
128        We need this information to compute the next wait time.
129        """
130
131        self._update_count: int = 0
132        """ Keep track of the number of times update() has been called. """
133
134        self._animation_path: str | None = animation_path
135        """ If specified, create a animation and write it to this location after the game completes. """
136
137        if (self._animation_path is not None):
138            if (os.path.splitext(self._animation_path)[-1] not in ANIMATION_EXTS):
139                raise ValueError(f"Animation path must have one of the following extensions {ANIMATION_EXTS}, found '{self._animation_path}'.")
140
141        self._animation_optimize: bool = animation_optimize
142        """ Optimize the animation output to reduce file size. """
143
144        self._animation_fps: int = max(MIN_ANIMATION_FPS, animation_fps)
145        """ The frame rate for the animation. """
146
147        self._animation_skip_frames: int = max(1, animation_skip_frames)
148        """
149        Skip this many frames between drawing animation frames.
150        This can help speed up animation creation by leaving out less important frames.
151        For example, this can be set to the number of agents to only draw frames after all agents have moved.
152        """
153
154        self._animation_frames: list[PIL.Image.Image] = []
155        """ The frames for the animation (one per call to update(). """
156
157        self._static_base_image: PIL.Image.Image | None = None
158        """
159        Cache an image that has all of the static (non-changing) elements (like walls) drawn.
160        This can be reused as the base image every time we draw an image.
161        """
162
163        # Only load sprites (and fonts) if we need them.
164        sprite_sheet = None
165        fonts = {}
166        if (self.requires_sprites() or (self._animation_path is not None)):
167            sprite_sheet = pacai.core.spritesheet.load(sprite_sheet_path)
168
169            for font_size in pacai.core.font.FontSize:
170                fonts[font_size] = PIL.ImageFont.truetype(font_path, int(sprite_sheet.height * font_size.value))
171
172        self._sprite_sheet: pacai.core.spritesheet.SpriteSheet | None = sprite_sheet
173        """ The sprite sheet to use for this UI. """
174
175        self._fonts: dict[pacai.core.font.FontSize, PIL.ImageFont.FreeTypeFont] = fonts
176        """ The available fonts indexed by size. """
177
178        self._image_cache: dict[int, PIL.Image.Image] = {}
179        """ Cache images (by game state turn count) to avoid redrawing images. """
180
181        self._highlights: dict[pacai.core.board.Position, float] = {}
182        """ The current set of board highlights. """
183
184    def update(self,
185            state: pacai.core.gamestate.GameState,
186            force_draw_image: bool = False,
187            board_highlights: list[pacai.core.board.Highlight] | None = None,
188            ) -> None:
189        """
190        Update the UI with the current state of the game.
191        This is the main entry point for the game into the UI.
192        """
193
194        self.wait_for_fps()
195
196        if (board_highlights is None):
197            board_highlights = []
198
199        for board_highlight in board_highlights:
200            intensity = board_highlight.get_float_intensity()
201            if (intensity is None):
202                self._highlights.pop(board_highlight.position, None)
203            else:
204                self._highlights[board_highlight.position] = intensity
205
206        if ((self._animation_path is not None) and (force_draw_image or (self._update_count % self._animation_skip_frames == 0))):
207            image = self.draw_image(state)
208            self._animation_frames.append(image)
209
210        self.draw(state)
211
212        self._update_count += 1
213
214    def game_start(self,
215            initial_state: pacai.core.gamestate.GameState,
216            board_highlights: list[pacai.core.board.Highlight] | None = None,
217            ) -> None:
218        """ Initialize the UI with the game's initial state. """
219
220        self.update(initial_state, board_highlights = board_highlights, force_draw_image = True)
221
222    def game_complete(self,
223            final_state: pacai.core.gamestate.GameState,
224            board_highlights: list[pacai.core.board.Highlight] | None = None,
225            ) -> None:
226        """ Update the UI with the game's final state. """
227
228        self.update(final_state, board_highlights = board_highlights, force_draw_image = True)
229
230        # Write the animation.
231        if ((self._animation_path is not None) and (len(self._animation_frames) > 0)):
232            ms_per_frame = int(1.0 / self._animation_fps * 1000.0)
233
234            options = {
235                'save_all': True,
236                'append_images': self._animation_frames,
237                'duration': ms_per_frame,
238                'loop': 0,
239                'optimize': False,
240                'minimize_size': False,
241            }
242
243            if (self._animation_optimize):
244                options['optimize'] = True
245                options['minimize_size'] = True
246
247            self._animation_frames[0].save(self._animation_path, None, **options)
248
249    def wait_for_fps(self) -> None:
250        """
251        Wait/Sleep for long enough to get close to the desired FPS.
252        Not all UIs will provide a real implementation for this method.
253        """
254
255        # No FPS limit is in place.
256        if (self._fps <= 0):
257            return
258
259        # This is the first wait request, we don't have enough information yet.
260        if (self._last_fps_wait is None):
261            self._last_fps_wait = edq.util.time.Timestamp.now()
262            return
263
264        last_time = self._last_fps_wait
265        now = edq.util.time.Timestamp.now()
266
267        duration = now.sub(last_time)
268
269        # Get the ideal number of milliseconds between frames.
270        ideal_time_between_frames_ms = 1000.0 / self._fps
271
272        # Get the wait time by comparing how long it has been since the last wait,
273        # with the ideal wait between frames.
274        wait_time_ms = ideal_time_between_frames_ms - duration.to_msecs()
275        if (wait_time_ms > 0):
276            self.sleep(int(wait_time_ms))
277
278        # Mark the time this method completed.
279        self._last_fps_wait = edq.util.time.Timestamp.now()
280
281    def requires_sprites(self) -> bool:
282        """ Check if this specific UI needs sprites or sprite sheets. """
283
284        return True
285
286    def sleep(self, sleep_time_ms: int) -> None:
287        """
288        Sleep for the specified number of ms.
289        This is in a method so children can override with any more UI-specific sleep procedures.
290        """
291
292        time.sleep(sleep_time_ms / 1000.0)
293
294    def close(self) -> None:
295        """ Close the UI and release all owned resources. """
296
297        if (self._user_input_device is not None):
298            self._user_input_device.close()
299
300    def get_user_inputs(self) -> list[pacai.core.action.Action]:
301        """
302        If a user input device is available,
303        get the inputs via UserInputDevice.get_inputs().
304        If no device is available, return an empty list.
305        """
306
307        if (self._user_input_device is None):
308            return []
309
310        return self._user_input_device.get_inputs()
311
312    def draw_image(self, state: pacai.core.gamestate.GameState, **kwargs: typing.Any) -> PIL.Image.Image:
313        """
314        Visualize the state of the game as an image.
315        This method is typically used for rendering the game to an animation.
316        each call to this method is one frame in the animation.
317        """
318
319        if (self._sprite_sheet is None):
320            raise ValueError("Cannot draw images without a sprite sheet.")
321
322        # First, check the cache for the image.
323        if (state.turn_count in self._image_cache):
324            return self._image_cache[state.turn_count]
325
326        image = self._get_static_image(state, **kwargs)
327
328        canvas = PIL.ImageDraw.Draw(image)
329
330        # Draw highlights.
331        for (position, base_intensity) in self._highlights.items():
332            start_coord = self._position_to_image_coords(position)
333            end_coord = self._position_to_image_coords(position.add(pacai.core.board.Position(1, 1)))
334
335            # Don't let the intensity go to zero.
336            intensity = 0.10 + (0.9 * base_intensity)
337
338            highlight_color = (
339                int(self._sprite_sheet.highlight[0] * intensity),
340                int(self._sprite_sheet.highlight[1] * intensity),
341                int(self._sprite_sheet.highlight[2] * intensity),
342            )
343
344            canvas.rectangle([start_coord, end_coord], fill = tuple(highlight_color))
345
346        # Draw non-agent (non-wall) markers.
347        for (marker, positions) in state.board._nonwall_objects.items():
348            if (marker.is_agent()):
349                continue
350
351            for position in positions:
352                if (state.skip_draw(marker, position, static = False)):
353                    continue
354
355                sprite = self._get_sprite(state, position, marker = marker, animation_key = ANIMATION_KEY)
356                self._place_sprite(position, sprite, image)
357
358        # Draw non-static text.
359        self._draw_position_text(state.get_nonstatic_text(), image)
360
361        # Draw agent markers.
362        for (marker, positions) in state.board._nonwall_objects.items():
363            if (not marker.is_agent()):
364                continue
365
366            for position in positions:
367                if (state.skip_draw(marker, position, static = False)):
368                    continue
369
370                last_action = state.get_last_agent_action(marker.get_agent_index())
371                sprite = self._get_sprite(state, position, marker = marker, action = last_action, animation_key = ANIMATION_KEY)
372                self._place_sprite(position, sprite, image)
373
374        # Draw the footer (usually the score).
375        footer_text = state.get_footer_text()
376        if (footer_text is not None):
377            (base_x, base_y) = self._position_to_image_coords(pacai.core.board.Position(state.board.height, 0))
378            self._draw_text(footer_text, base_x, base_y, canvas)
379
380        # Store this image in the cache.
381        self._image_cache[state.turn_count] = image
382
383        return image
384
385    def _get_font(self, size: pacai.core.font.FontSize) -> PIL.ImageFont.FreeTypeFont:
386        font = self._fonts.get(size, None)
387        if (font is None):
388            raise ValueError("Font has not been loaded.")
389
390        return font
391
392    def _get_static_image(self, state: pacai.core.gamestate.GameState, **kwargs: typing.Any) -> PIL.Image.Image:
393        """
394        Get the base image that only contains static objects.
395        This method will handle caching the base static image.
396        """
397
398        if (self._sprite_sheet is None):
399            raise ValueError("Cannot draw images without a sprite sheet.")
400
401        # Check the cache.
402        if (self._static_base_image is not None):
403            return self._static_base_image.copy()
404
405        # Height is +1 to leave room for the score.
406        size = (
407            state.board.width * self._sprite_sheet.width,
408            (state.board.height + 1) * self._sprite_sheet.height,
409        )
410
411        # Add in an alpha channel to the background.
412        background_color = list(self._sprite_sheet.background)
413        background_color.append(255)
414
415        image = PIL.Image.new('RGB', size, tuple(background_color))
416
417        # Draw wall markers.
418        for position in state.board.get_walls():
419            if (state.skip_draw(pacai.core.board.MARKER_WALL, position, static = True)):
420                continue
421
422            adjacency = state.board.get_adjacent_walls(position)
423            sprite = self._get_sprite(state, position, marker = pacai.core.board.MARKER_WALL, adjacency = adjacency, animation_key = ANIMATION_KEY)
424            self._place_sprite(position, sprite, image)
425
426        # Draw an additional static markers.
427        for position in state.get_static_positions():
428            for marker in state.board.get(position):
429                if (state.skip_draw(marker, position, static = True)):
430                    continue
431
432                sprite = self._get_sprite(state, position, marker = marker, animation_key = ANIMATION_KEY)
433                self._place_sprite(position, sprite, image)
434
435        # Draw static text.
436        self._draw_position_text(state.get_static_text(), image)
437
438        # Cache the image.
439        self._static_base_image = image.copy()
440
441        return image
442
443    def _draw_position_text(self, board_texts: list[pacai.core.font.BoardText], image: PIL.Image.Image) -> None:
444        """ Draw text on a board position. """
445
446        if (len(board_texts) == 0):
447            return
448
449        canvas = PIL.ImageDraw.Draw(image)
450        for board_text in board_texts:
451            # Base positions start in the upper left.
452            (base_x, base_y) = self._position_to_image_coords(board_text.position)
453
454            self._draw_text(board_text, base_x, base_y, canvas)
455
456    def _draw_text(self,
457            text: pacai.core.font.Text,
458            base_x: int, base_y: int,
459            canvas: PIL.ImageDraw.ImageDraw,
460            ) -> None:
461        """ Draw text to the board. """
462
463        if (self._sprite_sheet is None):
464            raise ValueError("Cannot draw text without a sprite sheet.")
465
466        # Compute alignment offsets.
467        vertical_offset = self._sprite_sheet.height * text.vertical_align.value
468        horizontal_offset = self._sprite_sheet.width * text.horizontal_align.value
469
470        y = base_y + vertical_offset
471        x = base_x + horizontal_offset
472
473        color = text.color
474        if (color is None):
475            color = self._sprite_sheet.text
476
477        canvas.text((x, y), text.text, color,
478                self._get_font(text.size),
479                anchor = text.anchor,
480                align = 'center')
481
482    def _get_sprite(self, state: pacai.core.gamestate.GameState, position: pacai.core.board.Position, **kwargs: typing.Any) -> PIL.Image.Image:
483        """ Get the requested sprite. """
484
485        if (self._sprite_sheet is None):
486            raise ValueError("Sprites are not loaded in this UI.")
487
488        return state.sprite_lookup(self._sprite_sheet, position, **kwargs)
489
490    def _place_sprite(self, position: pacai.core.board.Position, sprite: PIL.Image.Image, image: PIL.Image.Image) -> None:
491        image_coordinates = self._position_to_image_coords(position)
492
493        # Overlay the sprite onto the image.
494        # Note that the same image is used as the mask, since sprites will usually have alpha channels
495        # (so the transparent parts will not get drawn).
496        image.paste(sprite, image_coordinates, sprite)
497
498    def _position_to_image_coords(self, position: pacai.core.board.Position) -> tuple[int, int]:
499        """
500        Get the image coordinates (in pixels) for this position.
501        Returns: (x, y).
502        """
503
504        if (self._sprite_sheet is None):
505            raise ValueError("Sprites are not loaded.")
506
507        return self._sprite_sheet.position_to_pixels(position)
508
509    @abc.abstractmethod
510    def draw(self, state: pacai.core.gamestate.GameState, **kwargs: typing.Any) -> None:
511        """
512        Visualize the state of the game to the UI.
513        This is the typically the main override point for children.
514        Note that how this method visualizes the game completely unrelated
515        to how the draw_image() method works.
516        draw() will render to whatever the specific UI for the child class is,
517        while draw_image() specifically creates an image which will be used for animations.
518        If the child UI is also image-based than it can leverage draw_image(),
519        but there is no requirement to do that.
520        """
521
522def set_cli_args(parser: argparse.ArgumentParser) -> argparse.ArgumentParser:
523    """
524    Set common CLI arguments.
525    This is a sibling to init_from_args(), as the arguments set here can be interpreted there.
526    """
527
528    parser.add_argument('--ui', dest = 'ui',
529            action = 'store', type = str, default = pacai.util.alias.UI_WEB.short,
530            help = ('Set the UI/graphics to use (default: %(default)s).'
531                    + ' Builtin options:'
532                    + f' `{pacai.util.alias.UI_NULL.short}` (`{pacai.util.alias.UI_NULL.long}`)'
533                    +       ' -- Do not show any ui/graphics (best if you want to run fast and just need the result),'
534                    + f' `{pacai.util.alias.UI_STDIO.short}` (`{pacai.util.alias.UI_STDIO.long}`)'
535                    +       ' -- Use stdin/stdout from the terminal,'
536                    + f' `{pacai.util.alias.UI_TK.short}` (`{pacai.util.alias.UI_TK.long}`)'
537                    +       ' -- Use Tk/tkinter (must already be installed) to open a window,'
538                    + f' `{pacai.util.alias.UI_WEB.short}` (`{pacai.util.alias.UI_WEB.long}`)'
539                    +       ' -- Launch a browser window (default).'))
540
541    parser.add_argument('--show-training-ui', dest = 'show_training_ui',
542            action = 'store_true', default = False,
543            help = 'Show the specified UI (--ui) for training epochs/games. Otherwise, a null UI will be used (default: %(default)s).')
544
545    parser.add_argument('--fps', dest = 'fps',
546            action = 'store', type = int, default = DEFAULT_FPS,
547            help = ('Set the visual speed (frames per second) for UIs (default: %(default)s).'
548                    + ' Lower values are slower, and higher values are faster.'))
549
550    parser.add_argument('--animation-path', dest = 'animation_path',
551            action = 'store', type = str, default = None,
552            help = ('If specified, store an animated recording of the game at the specified location.'
553                    + f" This path must have one of the following extensions: {ANIMATION_EXTS}."))
554
555    parser.add_argument('--animation-fps', dest = 'animation_fps',
556            action = 'store', type = int, default = DEFAULT_ANIMATION_FPS,
557            help = 'Set the fps of the animation (default: %(default)s).')
558
559    parser.add_argument('--animation-skip-frames', dest = 'animation_skip_frames',
560            action = 'store', type = int, default = DEFAULT_ANIMATION_SKIP_FRAMES,
561            help = ('Only include every X frames in the animation.'
562                    + ' The default (1) means that every frame will be included.'
563                    + ' Using higher values can reduce the animations size and processing time'
564                    + ' (since there are fewer frames).'))
565
566    parser.add_argument('--animation-optimize', dest = 'animation_optimize',
567            action = 'store_true', default = DEFAULT_ANIMATION_OPTIMIZE,
568            help = 'Optimize the animation to reduce file size (will take longer) (default: %(default)s).')
569
570    return parser
571
572def init_from_args(
573        args: argparse.Namespace,
574        num_uis: int = 0,
575        null_out_uis: int = 0,
576        additional_args: dict | None = None,
577        ) -> argparse.Namespace:
578    """
579    Take in args from a parser that was passed to set_cli_args(),
580    and initialize the proper components.
581    Constructed UIs will be placed `args._uis`.
582    If `num_uis` is not provided (or <= 0),
583    then `args.num_games` + `args.num_training` will be used.
584    If `null_out_uis` is > 0, then at most that number of UIs (starting at the beginning)
585    will be converted to null UIs.
586    This will not change the total number of UIs, just null out the first number of UIs.
587    """
588
589    ui_args = {
590        'fps': args.fps,
591        'animation_path': args.animation_path,
592        'animation_fps': args.animation_fps,
593        'animation_skip_frames': args.animation_skip_frames,
594        'animation_optimize': args.animation_optimize,
595    }
596
597    if (additional_args is not None):
598        ui_args.update(additional_args)
599
600    if (num_uis <= 0):
601        num_uis = args.num_games + args.num_training
602
603    uis = []
604    for i in range(num_uis):
605        ui_name = args.ui
606        if (i < null_out_uis):
607            ui_name = pacai.util.alias.UI_NULL.long
608
609        uis.append(pacai.util.reflection.new_object(ui_name, **ui_args))
610
611    setattr(args, '_uis', uis)
612
613    return args