src/game/game.py

from collections import OrderedDict
from typing import Tuple, Union

import numpy as np
import pygame
from PIL import Image

from .dataset_env import DatasetEnv
from .keymap import Keymap
from .play_env import PlayEnv


class Game:
    def __init__(
        self,
        play_env: Union[PlayEnv, DatasetEnv],
        keymap: Keymap,
        size: Tuple[int, int],
        fps: int,
        verbose: bool,
    ) -> None:
        self.env = play_env
        self.keymap = keymap
        self.height, self.width = size
        self.fps = fps
        self.verbose = verbose
        self.keymap = OrderedDict()
        for keys, act in sorted(keymap.items(), key=lambda keys_act: -len(keys_act[0])):
            self.keymap[keys] = act

        print("\nControls (general):\n")
        print("⏎ : reset env")
        print(". : pause/unpause")
        print("e : step-by-step (when paused)")
        self.env.print_controls()
        print("\n")

    def run(self) -> None:
        pygame.init()

        header_height = 150 if self.verbose else 0
        font_size = 16
        screen = pygame.display.set_mode((self.width, self.height + header_height))
        clock = pygame.time.Clock()
        font = pygame.font.SysFont("mono", font_size)
        header_rect = pygame.Rect(0, 0, self.width, header_height)

        def clear_header():
            pygame.draw.rect(screen, pygame.Color("black"), header_rect)
            pygame.draw.rect(screen, pygame.Color("white"), header_rect, 1)

        def draw_text(text, idx_line, idx_column, num_cols):
            pos = (5 + idx_column * int(self.width // num_cols), 5 + idx_line * font_size)
            assert (0 <= pos[0] <= self.width) and (0 <= pos[1] <= header_height)
            screen.blit(font.render(text, True, pygame.Color("white")), pos)

        def draw_game(obs):
            assert obs.ndim == 4 and obs.size(0) == 1
            img = Image.fromarray(obs[0].add(1).div(2).mul(255).byte().permute(1, 2, 0).cpu().numpy())
            pygame_image = np.array(img.resize((self.width, self.height), resample=Image.NEAREST)).transpose((1, 0, 2))
            surface = pygame.surfarray.make_surface(pygame_image)
            screen.blit(surface, (0, header_height))

        def reset():
            nonlocal obs, info, do_reset, ep_return, ep_length
            obs, info = self.env.reset()
            do_reset = False
            ep_return = 0
            ep_length = 0

        obs, info, do_reset, ep_return, ep_length = (None,) * 5

        reset()
        do_wait = False
        should_stop = False

        while not should_stop:
            do_one_step = False
            action = 0  # noop
            pygame.event.pump()

            for event in pygame.event.get():
                if event.type == pygame.QUIT:
                    should_stop = True

                if event.type != pygame.KEYDOWN:
                    continue

                if event.key == pygame.K_RETURN:
                    do_reset = True

                if event.key == pygame.K_PERIOD:
                    do_wait = not do_wait
                    print("Game paused." if do_wait else "Game resumed.")

                if event.key == pygame.K_e:
                    do_one_step = True

                if event.key == pygame.K_m:
                    do_reset = self.env.next_mode()

                if event.key == pygame.K_UP:
                    do_reset = self.env.next_axis_1()

                if event.key == pygame.K_DOWN:
                    do_reset = self.env.prev_axis_1()

                if event.key == pygame.K_RIGHT:
                    do_reset = self.env.next_axis_2()

                if event.key == pygame.K_LEFT:
                    do_reset = self.env.prev_axis_2()

            if action == 0:
                pressed = pygame.key.get_pressed()
                for keys, action in self.keymap.items():
                    if all([pressed[key] for key in keys]):
                        break
                else:
                    action = 0

            if do_reset:
                reset()

            if do_wait and not do_one_step:
                continue

            next_obs, rew, end, trunc, info = self.env.step(action)

            ep_return += rew.item()
            ep_length += 1

            draw_game(obs)

            if self.verbose and info is not None:
                clear_header()
                assert isinstance(info, dict) and "header" in info
                header = info["header"]
                num_cols = len(header)
                for j, col in enumerate(header):
                    for i, row in enumerate(col):
                        draw_text(row, idx_line=i, idx_column=j, num_cols=num_cols)

            pygame.display.flip()  # update screen
            clock.tick(self.fps)  # ensures game maintains the given frame rate

            if end or trunc:
                reset()

            else:
                obs = next_obs

        pygame.quit()