game/pygame_base.py

from abc import ABC, abstractmethod
import logging
import random
import threading
import time
import pygame

from utils.config import Config
from utils.game_utils import capture

config = Config()


class PygameBase(ABC):
    @abstractmethod
    def __init__(self, output_dir, name):
        """Initialize the game, including setting up the game board, player, and environment."""
        
        self.output_dir = output_dir
        self.last_frame_base64 = None
        self.last_frame_path = None
        self.clock = pygame.time.Clock()
        
        # Load the level config, should be in Dict format
        self.level_config = config.level_config
        self.set_level_config(self.level_config)
        
        
        # pygame.mixer.pre_init(44100, -16, 2, 4096)
        pygame.init()
        
        self.over = False
        self.score = 0
        
        self.new_action_event = threading.Event()  # 用于通知有新动作
        self.new_action_event.set()
        
        self.history_frame_base64 = None
        self.history_action = None
        self.history_action_info = None
        self.history_valid_action = None
        self.action_in_sample_frames = None
        self.invalid_action_count = 0
        
        self.game_frames = []
        
        self.sample_frames = 3
        
        pygame.display.set_caption(name)
        

    @abstractmethod
    def step(self, action, dt=None):
        """
        Perform one step in the game based on the action taken.
        :param action: Action to be performed (could be direction, button press, etc.)
        :return: (done: bool, info: Any) Whether the game is over and any additional info, such as score.
        """
        pass
    
    
    @abstractmethod
    def human_mode_action(self, event):
        """
        Get the action from human player.
        :param event: pygame event
        :return: action: str
        """
        pass
    
    def get_score(self):
        """
        Retrieve the current score.
        Returns:
            int: The current score.
        """
        return self.score
    
    def get_game_info(self):
        #TODO: history should be managed by agent, not game
        info = {
            # Last step info
            "history_action": self.history_action,
            "history_frame_base64": self.history_frame_base64,
            "history_action_info": self.history_action_info,
            
            # Current step info
            "last_frame_base64": self.last_frame_base64,
            "last_frame_path": self.last_frame_path,
        }
        return info
    
    def init_run(self):
        
        self.step("None", 0)
        
        filepath, encoded_string = capture(self.screen, self.output_dir)
        self.last_frame_path=filepath
        self.last_frame_base64 = encoded_string
        self.game_frames.append(filepath)
        self.history_frame_base64 = self.last_frame_base64
        self.history_action = "None"
        self.history_action_info = "Game Initialization."
        self.history_valid_action = "None"
    
    def run(self, input_thread_method, human_mode=False):
        print("--------------------------------RUNNING---------------------------------------------")
        thread_event = threading.Event()
        if not human_mode:
            input_thread = threading.Thread(target=input_thread_method, args=(thread_event,), daemon=True)
            input_thread.start()
            
        self.init_run()
        
        waiting_time = 0
        invalid_action_count_temp = 0
        self.new_action_event.clear()  # 清除事件，准备接收下一个动作
        frame = 0
        frame_flag = True
        
        print("self.sample_frames", self.sample_frames)
        
        while not self.over:
            
            # dt = 0.033
            dt = 0.01
            self.clock.tick(config.FPS)
            
            if not frame_flag:
                frame += 1
                if frame != self.sample_frames:
                    step_action = self.history_valid_action
                    if self.action_in_sample_frames is not None:
                        step_action = self.action_in_sample_frames
                        
                    done, info = self.step(step_action, dt)
                    logging.info(info)
                    print(info)
                    print(f"frames {frame} step {step_action}")
                    if done:
                        self.over = True
                        print(info)
                        break
                        
                    self.history_action_info = info
                    filepath, encoded_string = capture(self.screen, self.output_dir)
                    self.last_frame_path=filepath
                    self.last_frame_base64 = encoded_string
                    self.game_frames.append(filepath)
                    continue
                else:
                    print(f"frames {frame} == {self.sample_frames}")
                    frame = 0   
                    if not human_mode:
                        self.new_action_event.clear()  # 清除事件，准备接收下一个动作 
                    frame_flag = True

            if not human_mode:
                if not input_thread.is_alive():
                    print("Restarting input thread...")
                    thread_event = threading.Event()
                    input_thread = threading.Thread(target=input_thread_method, args=(thread_event,), daemon=True)
                    time.sleep(5)
                    input_thread.start()
            
            # human_mode
            # if not self.new_action_event.is_set():
            action = None
            for event in pygame.event.get():
                action = self.human_mode_action(event)
                # print(self)
                if action is None or action not in self.valid_actions:
                    continue
                print("pygame base Action: ", action)
                done, info = self.step(action, dt)
                frame_flag = False
                # History
                self.history_frame_base64 = self.last_frame_base64
                self.history_action = action
                self.history_valid_action = action
                self.history_action_info = info
                filepath, encoded_string = capture(self.screen, self.output_dir)
                self.last_frame_path=filepath
                self.last_frame_base64 = encoded_string
                self.game_frames.append(filepath)
                if done:
                    run = False
                else:
                    print(self)
                    
                waiting_time = 0
            
            # 从队列中获取动作
            if self.new_action_event.is_set():
                waiting_time = 0
                action = self.current_action
                self.current_action = None
                
                if action in self.valid_actions:
                    # Valid action
                    done, info = self.step(action, dt)
                    
                    logging.info(info)
                    logging.info("---------------------------NEXT STEP--------------------------")
                    
                    print(info)
                    print("---------------------------NEXT STEP--------------------------")
                    
                    # History
                    self.history_frame_base64 = self.last_frame_base64
                    self.history_action = action
                    self.history_action_info = info
                    self.history_valid_action = action
                    
                    
                    filepath, encoded_string = capture(self.screen, self.output_dir)
                    self.last_frame_path=filepath
                    self.last_frame_base64 = encoded_string
                    self.game_frames.append(filepath)
                    
                    invalid_action_count_temp = 0
                    
                    frame_flag = False
                    
                    if done:
                        self.over = True
                else:
                    info = "Invalid action, which should be one of " + str(self.valid_actions)
                    print("action:", action,", info:", info)
                    logging.error(info)
                    invalid_action_count_temp += 1
                    self.invalid_action_count += 1
                    if invalid_action_count_temp == 3:
                        print("Invalid action for 3 times, take a random action.")
                        logging.error("Invalid action for 3 times, take a random action.")
                        action = random.choices(self.valid_actions, k=1)[0]
                        invalid_action_count_temp = 0
                        done, info = self.step(action, dt)
                        # History
                        self.history_frame_base64 = self.last_frame_base64
                        self.history_action = action
                        self.history_action_info = info
                        self.history_valid_action = action
                        frame_flag = False
                        filepath, encoded_string = capture(self.screen, self.output_dir)
                        self.last_frame_path=filepath
                        self.last_frame_base64 = encoded_string
                        self.game_frames.append(filepath)
                        
                        if done:
                            self.over = True
                        # action = input("Please enter your action (left, right, up, down or quit): ").strip().lower()
                    else:
                        # History
                        self.history_action = action
                        self.history_action_info = info
                        self.history_frame_base64 = self.last_frame_base64
                        filepath, encoded_string = capture(self.screen, self.output_dir)
                        self.last_frame_path=filepath
                        self.last_frame_base64 = encoded_string
                        self.game_frames.append(filepath)
                        self.new_action_event.clear()  # 清除事件，准备接收下一个动作     
            else:
                waiting_time += 1
                if waiting_time % 120 == 0:
                    print(time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time())), "waiting for action.")
                if waiting_time > 2000 * 30:
                    print(time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time())), "Waiting for action for 200s, restart.")
                    # kill input thread
                    thread_event.set()
                    input_thread.join()
                    waiting_time = 0
                    
                    thread_event = threading.Event()
                    input_thread = threading.Thread(target=input_thread_method, args=(thread_event,), daemon=True)
                    input_thread.start()

        
        if not human_mode and input_thread.is_alive():
            thread_event.set()   # 通知输入线程退出
            input_thread.join()  # 等待输入线程退出