import argparse import json import logging import os import random import re import signal import subprocess import sys import uuid from enum import Enum from subprocess import Popen import numpy as np import pygame import pygame_gui import requests import yaml from pygame import mixer from websockets.sync.client import connect from cooperative_cuisine import ROOT_DIR from cooperative_cuisine.action import ActionType, InterActionData, Action from cooperative_cuisine.argument_parser import create_gui_parser from cooperative_cuisine.game_server import ( CreateEnvironmentConfig, WebsocketMessage, PlayerRequestType, ) from cooperative_cuisine.items import ItemInfo from cooperative_cuisine.pygame_2d_vis.drawing import Visualizer, CacheFlags from cooperative_cuisine.pygame_2d_vis.game_colors import colors from cooperative_cuisine.server_results import PlayerInfo from cooperative_cuisine.state_representation import StateRepresentation from cooperative_cuisine.utils import ( setup_logging, load_config_files, ) from cooperative_cuisine.pygame_2d_vis.generate_layout_thumbnail import layout_thumbnail class MenuStates(Enum): """Enumeration of "Page" types in the 2D pygame vis.""" Start = "Start" Tutorial = "Tutorial" PreGame = "PreGame" Game = "Game" PostGame = "PostGame" End = "End" log = logging.getLogger(__name__) """The logger for this module.""" class PlayerKeySet: """Set of keyboard keys for controlling a player. First four keys are for movement. Order: Down, Up, Left, Right. 5th key is for interacting with counters. 6th key ist for picking up things or dropping them. """ def __init__( self, move_keys: list[pygame.key], interact_key: pygame.key, pickup_key: pygame.key, switch_key: pygame.key, players: list[str], joystick: int | None, ): """Creates a player key set which contains information about which keyboard keys control the player. Movement keys in the following order: Down, Up, Left, Right Args: move_keys: The keys which control this players movement in the following order: Down, Up, Left, Right. interact_key: The key to interact with objects in the game. pickup_key: The key to pick items up or put them down. switch_key: The key for switching through controllable players. players: The player indices which this keyset can control. joystick: number of joystick (later check if available) """ self.move_vectors: list[list[int]] = [[-1, 0], [1, 0], [0, -1], [0, 1]] self.key_to_movement: dict[pygame.key, list[int]] = { key: vec for (key, vec) in zip(move_keys, self.move_vectors) } self.move_keys: list[pygame.key] = move_keys self.interact_key: pygame.key = interact_key self.pickup_key: pygame.key = pickup_key self.switch_key: pygame.key = switch_key self.controlled_players: list[str] = players self.current_player: str = players[0] if players else "0" self.current_idx = 0 self.other_keyset: list[PlayerKeySet] = [] self.joystick = joystick def set_controlled_players(self, controlled_players: list[str]) -> None: """Sets the controlled players for this keyset. Args: controlled_players: The players controlled by this keyset. """ self.controlled_players = controlled_players self.current_player = self.controlled_players[0] self.current_idx = 0 def next_player(self) -> None: """Switches to the next player in the list of controlled players.""" self.current_idx = (self.current_idx + 1) % len(self.controlled_players) if self.other_keyset: for ok in self.other_keyset: if ok.current_idx == self.current_idx: self.next_player() return self.current_player = self.controlled_players[self.current_idx] def __repr__(self) -> str: return f"Keyset(current={self.current_player}, players={self.controlled_players}, joy={self.joystick})" class PyGameGUI: """Visualisation of the overcooked environment and reading keyboard inputs using pygame.""" def __init__( self, study_host: str, study_port: int, game_host: str, game_port: int, manager_ids: list[str], USE_AAAMBOS_AGENT: bool, debug: bool, do_study: bool, use_ssl: bool, ): self.USE_AAAMBOS_AGENT = USE_AAAMBOS_AGENT self.show_debug_elements = not do_study self.CONNECT_WITH_STUDY_SERVER = not debug pygame.init() pygame.display.set_icon( pygame.image.load(ROOT_DIR / "pygame_2d_vis" / "images" / "brain_icon.png") ) self.participant_id = uuid.uuid4().hex self.game_screen: pygame.Surface | None = None self.running = True self.key_sets: list[PlayerKeySet] = [] self.websockets = {} if self.CONNECT_WITH_STUDY_SERVER: self.request_url = ( f"http{'s' if use_ssl else ''}://{study_host}:{study_port}" ) else: self.request_url = f"http{'s' if use_ssl else ''}://{game_host}:{game_port}" self.manager_id = random.choice(manager_ids) with open(ROOT_DIR / "pygame_2d_vis" / "visualization.yaml", "r") as file: self.visualization_config = yaml.safe_load(file) self.language = self.visualization_config["Gui"]["language"] # self.fluent_bundle: FluentBundle = create_bundle( # self.visualization_config["Gui"]["language"], # ROOT_DIR # / "pygame_2d_vis" # / "locales" # / f"gui_messages_{self.language}.ftl", # ) self.FPS = self.visualization_config["GameWindow"]["FPS"] self.min_width = self.visualization_config["GameWindow"]["min_width"] self.min_height = self.visualization_config["GameWindow"]["min_height"] self.buttons_width = self.visualization_config["GameWindow"]["buttons_width"] self.buttons_height = self.visualization_config["GameWindow"]["buttons_height"] self.order_bar_height = self.visualization_config["GameWindow"][ "order_bar_height" ] ( self.window_width_fullscreen, self.window_height_fullscreen, ) = pygame.display.get_desktop_sizes()[0] # if ( # self.window_width_fullscreen >= 3840 # and self.window_height_fullscreen >= 2160 # ): # self.window_width_fullscreen /= 2 # self.window_height_fullscreen /= 2 self.game_width = 0 self.game_height = 0 self.window_width_windowed = self.min_width self.window_height_windowed = self.min_height self.kitchen_width = 1 self.kitchen_height = 1 self.kitchen_aspect_ratio = 1 self.images_path = ROOT_DIR / "pygame_gui" / "images" self.vis = Visualizer(self.visualization_config) self.last_score: float = 0 self.switch_score_color: bool = False self.count_frames_score_label: int = 0 self.fullscreen = False self.menu_state = MenuStates.Start self.manager: pygame_gui.UIManager self.sub_processes = [] self.layout_file_paths_dict = { p.name: p for p in (ROOT_DIR / "configs" / "layouts").rglob("*.layout") } self.layout_file_paths = sorted(list(self.layout_file_paths_dict.keys())) self.current_layout_idx = 0 self.selected_layout = "basic.layout" self.last_state: StateRepresentation self.player_info = {"0": {"name": "0"}} self.level_info = { "name": "Level", "recipe_graphs": [], "number_players": -1, "kitchen_size": (0, 0), } self.last_level = False self.beeped_once = False self.all_completed_meals = [] self.last_completed_meals = [] self.all_recipes_labels = [] self.last_recipes_labels = [] self.layout_surfaces = {} def setup_player_keys(self, players: list[str], number_key_sets=1, disjunct=False): # First four keys are for movement. Order: Down, Up, Left, Right. # 5th key is for interacting with counters. # 6th key ist for picking up things or dropping them. if number_key_sets: key_set1 = PlayerKeySet( move_keys=[pygame.K_a, pygame.K_d, pygame.K_w, pygame.K_s], interact_key=pygame.K_f, pickup_key=pygame.K_e, switch_key=pygame.K_SPACE, players=players, joystick=None, ) key_set2 = PlayerKeySet( move_keys=[pygame.K_LEFT, pygame.K_RIGHT, pygame.K_UP, pygame.K_DOWN], interact_key=pygame.K_i, pickup_key=pygame.K_o, switch_key=pygame.K_p, players=players, joystick=None, ) key_sets = [key_set1, key_set2] if self.joysticks: for idx, key in enumerate(self.joysticks.keys()): if idx >= len(key_sets): break key_sets[idx].joystick = key if disjunct: key_set1.set_controlled_players(players[::2]) key_set2.set_controlled_players(players[1::2]) elif number_key_sets > 1: key_set1.set_controlled_players(players) key_set2.set_controlled_players(players) key_set1.other_keyset = [key_set2] key_set2.other_keyset = [key_set1] key_set2.next_player() return key_sets[:number_key_sets] else: return [] def handle_keys(self): """Handles keyboard inputs. Sends action for the respective players. When a key is held down, every frame an action is sent in this function. """ keys = pygame.key.get_pressed() for key_set in self.key_sets: current_player_name = str(key_set.current_player) relevant_keys = [keys[k] for k in key_set.move_keys] if any(relevant_keys): move_vec = np.zeros(2) for idx, pressed in enumerate(relevant_keys): if pressed: move_vec += key_set.move_vectors[idx] if np.linalg.norm(move_vec) != 0: move_vec = move_vec / np.linalg.norm(move_vec) action = Action( current_player_name, ActionType.MOVEMENT.value, move_vec, duration=self.time_delta, ) self.send_action(action) def handle_joy_stick_input(self, joysticks: dict[int, pygame.joystick.Joystick]): """Handles joystick inputs for movement every frame Args: joysticks: list of joysticks """ # Axis 0: joy stick left: -1 = left, ~0 = center, 1 = right # Axis 1: joy stick left: -1 = up, ~0 = center, 1 = down # see control stuff here (at the end of the page): https://www.pygame.org/docs/ref/joystick.html for key_set in self.key_sets: current_player_name = str(key_set.current_player) # if a joystick is connected for current player if key_set.joystick in joysticks: # Usually axis run in pairs, up/down for one, and left/right for the other. Triggers count as axes. # You may want to take into account some tolerance to handle jitter, and # joystick drift may keep the joystick from centering at 0 or using the full range of position values. tolerance_threshold = 0.22 # axis 0 = joy stick left --> left & right axis_left_right = joysticks[key_set.joystick].get_axis(0) axis_up_down = joysticks[key_set.joystick].get_axis(1) if np.linalg.norm([axis_left_right, axis_up_down]) > tolerance_threshold: move_vec = np.array([axis_left_right, axis_up_down]) # if np.linalg.norm(move_vec) != 0: # move_vec = move_vec / np.linalg.norm(move_vec) action = Action( current_player_name, ActionType.MOVEMENT.value, move_vec, duration=self.time_delta, ) self.send_action(action) def handle_key_event(self, event: pygame.event.Event): """Handles key events for the pickup and interaction keys. Pickup is a single action, for interaction keydown and keyup is necessary, because the player has to be able to hold the key down. Args: event: Pygame event for extracting the key action. """ for key_set in self.key_sets: current_player_name = str(key_set.current_player) if event.key == key_set.pickup_key and event.type == pygame.KEYDOWN: action = Action(current_player_name, ActionType.PICK_UP_DROP, None) self.send_action(action) if event.key == key_set.interact_key: if event.type == pygame.KEYDOWN: action = Action( current_player_name, ActionType.INTERACT, InterActionData.START ) self.send_action(action) elif event.type == pygame.KEYUP: action = Action( current_player_name, ActionType.INTERACT, InterActionData.STOP ) self.send_action(action) if event.key == key_set.switch_key: if event.type == pygame.KEYDOWN: key_set.next_player() def handle_joy_stick_event( self, event: pygame.event.Event, joysticks: dict[int, pygame.joystick.Joystick] ): """Handles joy stick events for the pickup and interaction keys. Pickup is a single action, for interaction buttondown and buttonup is necessary, because the player has to be able to hold the button down. Args: event: Pygame event for extracting the button action. joysticks: list of joysticks """ for key_set in self.key_sets: current_player_name = str(key_set.current_player) # if a joystick is connected for current player if key_set.joystick in joysticks: # pickup = Button A <-> 0 if ( joysticks[key_set.joystick].get_button(0) and event.type == pygame.JOYBUTTONDOWN ): action = Action(current_player_name, ActionType.PICK_UP_DROP, None) self.send_action(action) # interact = Button X <-> 2 if ( joysticks[key_set.joystick].get_button(2) and event.type == pygame.JOYBUTTONDOWN ): action = Action( current_player_name, ActionType.INTERACT, InterActionData.START ) self.send_action(action) # stop interaction if last pressed button was X <-> 2 if event.button == 2 and event.type == pygame.JOYBUTTONUP: action = Action( current_player_name, ActionType.INTERACT, InterActionData.STOP ) self.send_action(action) # switch button Y <-> 3 if joysticks[key_set.joystick].get_button(3): if event.type == pygame.JOYBUTTONDOWN: key_set.next_player() def set_window_size(self): """Sets the window size based on fullscreen or not.""" if self.fullscreen: flags = pygame.FULLSCREEN self.window_width = self.window_width_fullscreen self.window_height = self.window_height_fullscreen else: flags = 0 self.window_width = self.window_width_windowed self.window_height = self.window_height_windowed self.screen_margin = self.visualization_config["GameWindow"][ "screen_margin_proportion" ] * min(self.window_width, self.window_height) self.main_window = pygame.display.set_mode( ( self.window_width, self.window_height, ), flags=flags, ) def set_game_size(self, max_width: float = None, max_height: float = None): """Sets the game size based on the kitchen size and the current window size. Args: max_width: Maximum width of the game screen. max_height: Maximum height of the game screen. """ if max_width is None: max_width = self.window_width - (2 * self.screen_margin) if max_height is None: max_height = self.window_height - (2 * self.screen_margin) self.kitchen_aspect_ratio = self.kitchen_height / self.kitchen_width if self.kitchen_width > self.kitchen_height: self.game_width = max_width self.game_height = self.game_width * self.kitchen_aspect_ratio if self.game_height > max_height: self.game_height = max_height self.game_width = self.game_height / self.kitchen_aspect_ratio else: self.game_height = max_height self.game_width = self.game_height / self.kitchen_aspect_ratio if self.game_width > max_width: self.game_width = max_width self.game_height = self.game_width * self.kitchen_aspect_ratio self.grid_size = int(self.game_width / self.kitchen_width) self.game_width = max(self.game_width, 100) self.game_height = max(self.game_height, 100) self.grid_size = max(self.grid_size, 1) self.vis.set_grid_size(self.grid_size) residual_x = self.game_width - (self.kitchen_width * self.grid_size) residual_y = self.game_height - (self.kitchen_height * self.grid_size) self.game_width -= residual_x self.game_height -= residual_y self.game_screen = pygame.Surface( ( self.game_width, self.game_height, ) ) def init_ui_elements(self): """Creates all UI elements. Creates lists of which elements belong on which screen.""" self.manager = pygame_gui.UIManager( (self.window_width, self.window_height), starting_language=self.language, translation_directory_paths=[ROOT_DIR / "pygame_2d_vis" / "locales"], ) self.manager.get_theme().load_theme( ROOT_DIR / "pygame_2d_vis" / "gui_theme.json" ) self.elements_margin = self.window_height * 0.02 ######################################################################## # All screens ######################################################################## fullscreen_button_rect = pygame.Rect( (0, 0), (self.buttons_width * 0.7, self.buttons_height) ) fullscreen_button_rect.topright = (-self.buttons_width, 0) self.fullscreen_button = pygame_gui.elements.UIButton( relative_rect=fullscreen_button_rect, text="translations.fullscreen", manager=self.manager, object_id="#fullscreen_button", anchors={"right": "right", "top": "top"}, ) rect = pygame.Rect((0, 0), (self.buttons_width, self.buttons_height)) rect.topright = (0, 0) self.quit_button = pygame_gui.elements.UIButton( relative_rect=rect, text="translations.quit_game", manager=self.manager, object_id="#quit_button", anchors={"right": "right", "top": "top"}, ) ######################################################################## # Start screen ######################################################################## self.start_button = pygame_gui.elements.UIButton( relative_rect=pygame.Rect( (0, 0), (self.buttons_width, self.buttons_height) ), text="translations.start_game", text_kwargs={}, manager=self.manager, anchors={"center": "center"}, object_id="#start_button", ) if self.visualization_config["Gui"]["press_button_to_continue"]: img = pygame.image.load( ROOT_DIR / "pygame_2d_vis" / "gui_images" / f"continue_{self.language}.png" ).convert_alpha() image_rect = img.get_rect() img_width = self.buttons_width * 1.5 img_height = img_width * (image_rect.height / image_rect.width) new_dims = (img_width, img_height) img = pygame.transform.smoothscale(img, new_dims) image_rect = img.get_rect() image_rect.centery += 80 self.press_a_image = pygame_gui.elements.UIImage( image_rect, img, manager=self.manager, anchors={"centerx": "centerx", "centery": "centery"}, ) else: self.press_a_image = None # # self.press_a_image.set_dimensions(new_dims) # if not self.CONNECT_WITH_STUDY_SERVER: # assert len(self.layout_file_paths) != 0, "No layout files." # dropdown_width, dropdown_height = 200, 40 # self.layout_selection = pygame_gui.elements.UIDropDownMenu( # relative_rect=pygame.Rect( # ( # 0, # 0, # ), # (dropdown_width, dropdown_height), # ), # manager=self.manager, # options_list=self.layout_file_paths, # starting_option="basic.layout" # if "basic.layout" in self.layout_file_paths # else random.choice(self.layout_file_paths), # ) player_selection_rect = pygame.Rect( (0, 0), ( self.window_width * 0.9, (self.window_height // 4), ), ) player_selection_rect.bottom = -self.elements_margin self.player_selection_container = pygame_gui.elements.UIPanel( player_selection_rect, manager=self.manager, object_id="#players", anchors={"bottom": "bottom", "centerx": "centerx"}, ) rect = pygame.Rect((0, 0), (190, 50)) self.multiple_keysets_button = pygame_gui.elements.UIButton( relative_rect=rect, manager=self.manager, container=self.player_selection_container, text="not set", anchors={"centerx": "centerx", "centery": "centery"}, object_id="#multiple_keysets_button", ) split_players_button_rect = pygame.Rect((0, 0), (190, 50)) self.split_players_button = pygame_gui.elements.UIButton( relative_rect=split_players_button_rect, manager=self.manager, container=self.player_selection_container, text="not set", anchors={"centery": "centery", "left_target": self.multiple_keysets_button}, object_id="#split_players_button", ) rect = pygame.Rect( (0, 0), ( self.window_width * 0.6, self.player_selection_container.get_abs_rect().height * 0.35, ), ) self.player_number_container = pygame_gui.elements.UIPanel( relative_rect=rect, manager=self.manager, object_id="#players_players", container=self.player_selection_container, anchors={"top": "top", "centerx": "centerx"}, ) rect = pygame.Rect( (0, 0), ( self.window_width * 0.6, self.player_selection_container.get_abs_rect().height * 0.35, ), ) rect.bottom = 0 self.bot_number_container = pygame_gui.elements.UIPanel( relative_rect=rect, manager=self.manager, object_id="#players_bots", container=self.player_selection_container, anchors={"bottom": "bottom", "centerx": "centerx"}, ) number_players_rect = pygame.Rect((0, 0), (200, 200)) self.added_players_label = pygame_gui.elements.UILabel( number_players_rect, manager=self.manager, object_id="#number_players_label", container=self.player_number_container, text=f"not set: -", anchors={"center": "center"}, ) number_bots_rect = pygame.Rect((0, 0), (200, 200)) self.added_bots_label = pygame_gui.elements.UILabel( number_bots_rect, manager=self.manager, object_id="#number_bots_label", container=self.bot_number_container, text=f"not set: -", anchors={"center": "center"}, ) size = 50 self.add_human_player_button = pygame_gui.elements.UIButton( relative_rect=pygame.Rect((0, 0), (size, size)), text="+", manager=self.manager, object_id="#quantity_button", container=self.player_number_container, anchors={"left_target": self.added_bots_label, "centery": "centery"}, ) rect = pygame.Rect((0, 0), (size, size)) rect.right = 0 self.remove_human_button = pygame_gui.elements.UIButton( relative_rect=rect, text="-", manager=self.manager, object_id="#quantity_button", container=self.player_number_container, anchors={ "right": "right", "right_target": self.added_bots_label, "centery": "centery", }, ) self.add_bot_button = pygame_gui.elements.UIButton( relative_rect=pygame.Rect((0, 0), (size, size)), text="+", manager=self.manager, object_id="#quantity_button", container=self.bot_number_container, anchors={"left_target": self.added_bots_label, "centery": "centery"}, ) rect = pygame.Rect((0, 0), (size, size)) rect.right = 0 self.remove_bot_button = pygame_gui.elements.UIButton( relative_rect=rect, text="-", manager=self.manager, object_id="#quantity_button", container=self.bot_number_container, anchors={ "right": "right", "right_target": self.added_bots_label, "centery": "centery", }, ) self.layouts_width = self.buttons_width * 1.2 self.scroll_space_layouts = pygame_gui.elements.UIScrollingContainer( relative_rect=pygame.Rect((0, 0), (self.layouts_width, self.window_height)), manager=self.manager, anchors={"top": "top", "left": "left"}, ) self.setup_layout_selection() rect = pygame.Rect( (0, 0), ( self.window_width * 0.5, self.buttons_height, ), ) self.selected_layout_label = pygame_gui.elements.UILabel( text="translations.selected_layout", text_kwargs={"layout": "basic.layout"}, relative_rect=rect, manager=self.manager, object_id="#selected_layout", anchors={"centerx": "centerx", "bottom_target": self.start_button}, ) dummy_surf = pygame.Surface((100, 100)) self.preview_layout_image = pygame_gui.elements.UIImage( relative_rect=pygame.Rect( (0, 0), (100, 100), ), image_surface=dummy_surf, manager=self.manager, object_id="#preview_layout", anchors={"centerx": "centerx", "top_target": self.selected_layout_label}, ) ######################################################################## # Tutorial screen ######################################################################## button_rect = pygame.Rect((0, 0), (220, 80)) button_rect.bottom = -self.elements_margin self.continue_button = pygame_gui.elements.UIButton( relative_rect=button_rect, text="translations.continue", manager=self.manager, anchors={"centerx": "centerx", "bottom": "bottom"}, ) # TODO update on pregame button_rect = pygame.Rect((0, 0), (220, 80)) button_rect.bottom = -self.elements_margin button_rect.centerx += 250 self.continue_button_two = pygame_gui.elements.UIButton( relative_rect=button_rect, text="translations.continue", manager=self.manager, anchors={"centerx": "centerx", "bottom": "bottom"}, ) image = pygame.image.load( ROOT_DIR / "pygame_2d_vis" / "gui_images" / f"controls_{self.language}.png" ).convert_alpha() image_rect = image.get_rect() # img_width = self.window_width * 0.68 # img_height = img_width * (image_rect.height / image_rect.width) img_height = self.window_height * 0.95 img_width = img_height * (image_rect.width / image_rect.height) new_dims = (img_width, img_height) image = pygame.transform.smoothscale(image, new_dims) image_rect = image.get_rect() image_rect.left = self.elements_margin image_rect.top = self.elements_margin self.tutorial_image = pygame_gui.elements.UIImage( image_rect, image, manager=self.manager, anchors={"top": "top", "left": "left"}, ) arrow_img = pygame.image.load( ROOT_DIR / "pygame_2d_vis" / "gui_images" / f"try_{self.language}.png" ).convert_alpha() image_rect = arrow_img.get_rect() img_width = self.window_width * 0.2 img_height = img_width * (image_rect.height / image_rect.width) new_dims = (img_width, img_height) arrow_img = pygame.transform.smoothscale(arrow_img, new_dims) rect = arrow_img.get_rect() rect.left = self.window_width * 0.55 rect.top = self.window_height * 0.7 self.arrow_img = pygame_gui.elements.UIImage( relative_rect=rect, image_surface=arrow_img, manager=self.manager, anchors={ "left": "left", "top": "top", }, ) rect = pygame.Rect( (0, 0), (self.window_width * 0.25, self.window_height * 0.4), ) rect.right = -self.elements_margin self.tutorial_recipe_container = pygame_gui.elements.UIPanel( relative_rect=rect, manager=self.manager, object_id="#graph_container", anchors={"right": "right", "top": "top", "top_target": self.quit_button}, ) self.tutorial_recipe_graph_rect = pygame.Rect( (0, 0), (self.window_width * 0.25, self.window_height * 0.2) ) self.tutorial_recipe_graph_rect.bottom = 0 self.tutorial_graph_image = pygame_gui.elements.UIImage( relative_rect=self.tutorial_recipe_graph_rect, image_surface=pygame.Surface(self.tutorial_recipe_graph_rect.size), manager=self.manager, object_id="#recipe_graph", container=self.tutorial_recipe_container, anchors={"centerx": "centerx", "bottom": "bottom"}, ) r = pygame.Rect((0, 0), (self.window_width * 0.25, self.window_height * 0.05)) r.bottom = 0 text = pygame_gui.elements.UILabel( text="translations.salad_recipe", relative_rect=r, manager=self.manager, container=self.tutorial_recipe_container, object_id="#recipe_name", anchors={ "centerx": "centerx", "bottom": "bottom", "bottom_target": self.tutorial_graph_image, }, ) ######################################################################## # PreGame screen ######################################################################## rect = pygame.Rect( (0, 0), (self.window_width, 50), ) rect.top = 20 self.level_name_label = pygame_gui.elements.UILabel( text=f"not set", relative_rect=rect, manager=self.manager, object_id="#level_name", anchors={"centerx": "centerx", "top": "top"}, ) self.text_recipes_label = pygame_gui.elements.UILabel( text="translations.recipes_in_this_level", relative_rect=pygame.Rect( (0, 0), (self.window_width * 0.5, 50), ), manager=self.manager, anchors={"centerx": "centerx", "top_target": self.level_name_label}, ) self.scroll_height = ( self.continue_button.get_abs_rect().top - self.text_recipes_label.get_abs_rect().bottom ) self.scroll_space_recipes = pygame_gui.elements.UIScrollingContainer( relative_rect=pygame.Rect((0, 0), (self.window_width, self.scroll_height)), manager=self.manager, anchors={"centerx": "centerx", "top_target": self.text_recipes_label}, ) ######################################################################## # Game screen ######################################################################## self.orders_label = pygame_gui.elements.UILabel( text="translations.orders", relative_rect=pygame.Rect(0, 0, self.buttons_width, self.screen_margin), manager=self.manager, object_id="#orders_label", ) self.orders_container_width = ( self.window_width - (2 * self.buttons_width) - (self.buttons_width * 0.7) ) self.orders_image = pygame_gui.elements.UIImage( relative_rect=pygame.Rect( 0, 0, self.orders_container_width, self.screen_margin ), image_surface=pygame.Surface( (self.orders_container_width, self.screen_margin) ), manager=self.manager, object_id="#recipe_graph", anchors={"top": "top", "left_target": self.orders_label}, ) rect = pygame.Rect( (0, 0), (self.window_width * 0.3, self.buttons_height), ) rect.bottomleft = (0, 0) self.score_label = pygame_gui.elements.UILabel( text=f"Score not set", relative_rect=rect, manager=self.manager, object_id="#score_label", anchors={"bottom": "bottom", "left": "left"}, ) rect = pygame.Rect( (0, 0), (self.window_width * 0.4, self.buttons_height), ) rect.bottom = 0 self.timer_label = pygame_gui.elements.UILabel( text="GAMETIME not set", relative_rect=rect, manager=self.manager, object_id="#timer_label", anchors={"bottom": "bottom", "centerx": "centerx"}, ) rect = pygame.Rect( (0, 0), (self.window_width, self.screen_margin), ) rect.right = 20 self.wait_players_label = pygame_gui.elements.UILabel( text="translations.waiting_for_players", relative_rect=rect, manager=self.manager, object_id="#wait_players_label", anchors={"centery": "centery", "right": "right"}, ) rect = pygame.Rect((0, 0), (self.buttons_width, self.buttons_height)) rect.bottomright = (0, 0) self.return_to_main_button = pygame_gui.elements.UIButton( relative_rect=rect, text="translations.return_to_main_menu", manager=self.manager, object_id="#main_menu_button", anchors={"right": "right", "bottom": "bottom"}, ) ######################################################################## # PostGame screen ######################################################################## rect = pygame.Rect((0, 0), (220, 80)) rect.bottom = -self.elements_margin self.next_game_button = pygame_gui.elements.UIButton( relative_rect=rect, manager=self.manager, text="translations.next_game", anchors={"centerx": "centerx", "bottom": "bottom"}, object_id="#split_players_button", ) rect = pygame.Rect( (0, 0), (self.window_width, self.window_height * 0.07), ) self.score_conclusion = pygame_gui.elements.UILabel( text=f"not set", relative_rect=rect, manager=self.manager, object_id="#score_conclusion", anchors={"centerx": "centerx", "top_target": self.level_name_label}, ) self.completed_meals_text_label = pygame_gui.elements.UILabel( text="translations.completed_meals", relative_rect=pygame.Rect( (0, 0), (self.window_width, self.window_height * 0.05), ), manager=self.manager, object_id="#completed_meals_label", anchors={"centerx": "centerx", "top_target": self.score_conclusion}, ) scroll_height = ( self.continue_button.get_abs_rect().top - self.completed_meals_text_label.get_abs_rect().bottom - 10 ) self.scroll_width_completed_meals = self.window_width self.scroll_space_completed_meals = pygame_gui.elements.UIScrollingContainer( relative_rect=pygame.Rect((0, 0), (self.window_width, scroll_height)), manager=self.manager, anchors={ "centerx": "centerx", "top_target": self.completed_meals_text_label, }, ) rect = pygame.Rect((0, 0), (220, 80)) rect.bottom = -self.elements_margin self.finish_study_button = pygame_gui.elements.UIButton( relative_rect=rect, manager=self.manager, text="translations.finish_study", anchors={"centerx": "centerx", "bottom": "bottom"}, object_id="#split_players_button", ) ######################################################################## # End screen ######################################################################## rect = pygame.Rect(0, 0, self.window_width * 0.9, self.window_height * 0.4) final_text_container = pygame_gui.elements.UIPanel( relative_rect=rect, manager=self.manager, object_id="#graph_container", anchors={"center": "center"}, ) height = self.window_height * 0.1 rect = pygame.Rect((0, 0), (self.window_width * 0.9, height)) rect.centery -= height / 2 text1 = pygame_gui.elements.UILabel( text="translations.thank_you", relative_rect=rect, manager=self.manager, object_id="#score_label", container=final_text_container, anchors={"centery": "centery", "centerx": "centerx"}, ) rect = pygame.Rect((0, 0), (self.window_width * 0.9, self.window_height * 0.1)) text2 = pygame_gui.elements.UILabel( text="translations.end_screen_line1", relative_rect=rect, manager=self.manager, object_id="#score_label", container=final_text_container, anchors={"top_target": text1, "centerx": "centerx"}, ) text3 = pygame_gui.elements.UILabel( text="translations.end_screen_line2", relative_rect=rect, manager=self.manager, object_id="#score_label", container=final_text_container, anchors={"top_target": text2, "centerx": "centerx"}, ) ######################################################################## if self.show_debug_elements: self.start_screen_elements = [ self.start_button, self.quit_button, self.fullscreen_button, self.player_selection_container, self.bot_number_container, self.press_a_image, ] self.on_all_screens = [ self.fullscreen_button, self.quit_button, ] self.other_elements = [] else: self.start_screen_elements = [ self.start_button, self.press_a_image, self.fullscreen_button, ] self.on_all_screens = [] self.other_elements = [ self.player_selection_container, self.bot_number_container, self.press_a_image, self.quit_button, self.fullscreen_button, self.continue_button_two, self.scroll_space_layouts ] self.tutorial_screen_elements = [ self.tutorial_image, self.continue_button, self.tutorial_recipe_container, self.arrow_img, ] self.pregame_screen_elements = [ self.level_name_label, self.text_recipes_label, self.scroll_space_recipes, self.continue_button_two, ] self.game_screen_elements = [ self.orders_label, self.orders_image, self.score_label, self.timer_label, self.wait_players_label, ] if not self.CONNECT_WITH_STUDY_SERVER: self.start_screen_elements.append(self.scroll_space_layouts) self.start_screen_elements.append(self.selected_layout_label) self.game_screen_elements.append(self.return_to_main_button) else: self.other_elements.append(self.scroll_space_layouts) self.other_elements.append(self.return_to_main_button) self.other_elements.append(self.selected_layout_label) self.postgame_screen_elements = [ self.score_conclusion, self.scroll_space_completed_meals, self.level_name_label, self.next_game_button, self.finish_study_button, self.completed_meals_text_label, ] self.end_screen_elements = [ final_text_container, self.quit_button ] def show_screen_elements(self, elements: list[pygame_gui.core.UIElement]): """Hides all UI elements and shows the elements in the list and elements in self.on_all_screens. Args: elements: List of UI elements to show. """ all_elements = ( self.start_screen_elements + self.tutorial_screen_elements + self.pregame_screen_elements + self.game_screen_elements + self.postgame_screen_elements + self.end_screen_elements + self.on_all_screens + self.other_elements ) for element in all_elements: if element: element.hide() for element in elements + self.on_all_screens: if element: element.show() def setup_layout_selection(self): """Sets up the layout selection screen with the layout selection dropdown.""" added_layouts = [] layout_display_width = self.layouts_width layout_display_height = self.buttons_height * 3 overall_scroll_height = layout_display_height * len(self.layout_file_paths_dict) self.layouts_main_container = pygame_gui.elements.UIPanel( relative_rect=pygame.Rect( (0, 0), ( layout_display_width, overall_scroll_height, ), ), object_id="#graph_container", manager=self.manager, container=self.scroll_space_layouts, # anchors={"centerx": "centerx", "top": "top"}, ) self.scroll_space_layouts.set_scrollable_area_dimensions( (self.layouts_width, overall_scroll_height) ) with open(ROOT_DIR / "pygame_2d_vis" / "visualization.yaml", "r") as file: visualization_config = yaml.safe_load(file) vis = Visualizer(visualization_config) vis.cache_flags = CacheFlags.NONE vis.create_player_colors(10) with open(ROOT_DIR / "configs" / "item_info.yaml") as file: item_info = file.read() item_lookup = yaml.safe_load(item_info) for item_name in item_lookup: item_lookup[item_name] = ItemInfo(name=item_name, **item_lookup[item_name]) for item_name, item_info in item_lookup.items(): if item_info.equipment: item_info.equipment = item_lookup[item_info.equipment] self.layout_buttons = {} for idx, layout_name in enumerate( sorted(self.layout_file_paths_dict.keys(), key=lambda x: "000000" if x == "basic.layout" else x)): if idx == 0: anchors = {"centerx": "centerx", "top": "top"} else: anchors = { "centerx": "centerx", "top_target": added_layouts[idx - 1], } container = pygame_gui.elements.UIPanel( relative_rect=pygame.Rect( (0, 0), ( layout_display_width, layout_display_height, ), ), object_id="#graph_container", manager=self.manager, container=self.layouts_main_container, anchors=anchors, ) layout_path = self.layout_file_paths_dict[layout_name] max_size = layout_display_height * 0.9 factor = 6 layout_surface = layout_thumbnail(layout_path, vis, max_size * factor, item_lookup) layout_image = pygame_gui.elements.UIImage( relative_rect=pygame.transform.scale_by(layout_surface, 1 / factor).get_rect(), image_surface=layout_surface, manager=self.manager, container=container, anchors={"center": "center"}, ) self.layout_surfaces[layout_name] = layout_surface w, h = layout_image.get_abs_rect().size rect = pygame.Rect( (0, 0), (w + 2, h + 2), ) button = pygame_gui.elements.UIButton( text=layout_name.replace(".layout", ""), relative_rect=rect, manager=self.manager, container=container, object_id="#layout_button", anchors={"centerx": "centerx", "centery": "centery"}, ) self.layout_buttons[button] = layout_name added_layouts.append(container) def setup_tutorial_screen(self): """Updates the tutorial screen with the current tutorial image and the continue button.""" self.show_screen_elements(self.tutorial_screen_elements) self.set_game_size( max_height=self.window_height * 0.4, max_width=self.window_width * 0.3, ) self.game_center = ( self.window_width - self.game_width / 2 - (self.window_height * 0.04), self.window_height - self.game_height / 2 - (self.window_height * 0.04), ) width, height = self.tutorial_recipe_graph_rect.size tutorial_graph_surface = pygame.Surface( self.tutorial_recipe_graph_rect.size, flags=pygame.SRCALPHA ) self.vis.draw_recipe_image( tutorial_graph_surface, self.level_info["recipe_graphs"][0], width, height, node_size=self.window_height / 18, ) self.tutorial_graph_image.set_image(tutorial_graph_surface) def update_screen_elements(self): """Shows and hides the UI elements based on the current menu state.""" self.vis.invalidate_surface_cache() match self.menu_state: case MenuStates.Start: self.show_screen_elements(self.start_screen_elements) if self.CONNECT_WITH_STUDY_SERVER: self.bot_number_container.hide() # else: # self.setup_layout_selection() self.update_selection_elements() case MenuStates.Tutorial: self.setup_tutorial_screen() case MenuStates.PreGame: self.init_ui_elements() self.show_screen_elements(self.pregame_screen_elements) self.update_pregame_screen() case MenuStates.Game: self.show_screen_elements(self.game_screen_elements) # print(self.grid_size, self.window_width, self.window_height) case MenuStates.PostGame: self.init_ui_elements() self.show_screen_elements(self.postgame_screen_elements) self.update_post_game_screen(self.last_state) if self.last_level: self.next_game_button.hide() self.finish_study_button.show() else: self.next_game_button.show() self.finish_study_button.hide() case MenuStates.End: self.show_screen_elements(self.end_screen_elements) def draw_main_window(self): """Main draw function. Draws the main window and updates the UI. Draws the game screen in Game and ControllerTutorial screen.""" self.main_window.fill( colors[self.visualization_config["GameWindow"]["background_color"]] ) if self.menu_state == MenuStates.Tutorial: self.draw_game_screen_frame(tutorial=True) elif self.menu_state == MenuStates.Game: self.draw_game_screen_frame(tutorial=False) self.manager.draw_ui(self.main_window) self.manager.update(self.time_delta) pygame.display.flip() def update_post_game_screen(self, state: dict): """Updates the post game screen with the final score and the completed meals. Args: state: The game state returned by the environment, containing served meals and score. """ score = state["score"] self.score_conclusion.set_text( "translations.final_score", text_kwargs={"score": str(score)} ) self.level_name_label.set_text( "translations.completed_level", text_kwargs={"level": str(self.level_info["name"])}, ) served_meals = state["served_meals"] row_height = self.window_height * 0.07 container_width = self.scroll_width_completed_meals container_height = len(served_meals) * row_height main_container = pygame_gui.elements.UIPanel( relative_rect=pygame.Rect( (0, 0), ( container_width, container_height, ), ), object_id="#graph_container", manager=self.manager, container=self.scroll_space_completed_meals, ) last_completed_meals = [] for idx, (player, meal) in enumerate(served_meals): if idx == 0: anchors = {"centerx": "centerx", "top": "top"} else: anchors = { "centerx": "centerx", "top_target": last_completed_meals[idx - 1], } rect = pygame.Rect( (0, 0), ( container_width / 2, row_height, ), ) container = pygame_gui.elements.UIPanel( relative_rect=rect, object_id="#graph_container", manager=self.manager, container=main_container, anchors=anchors, ) meal_surface = pygame.Surface( (row_height, row_height), flags=pygame.SRCALPHA ) self.vis.draw_item( pos=np.array([row_height / 2, row_height / 2]), item={"type": "Plate"}, plate=True, screen=meal_surface, grid_size=row_height, ) meal_name = meal.split("(")[0] self.vis.draw_item( pos=np.array([row_height / 2, row_height / 2]), item={"type": meal_name}, plate=True, screen=meal_surface, grid_size=row_height, ) rect = meal_surface.get_rect() rect.center = (rect.center[0] - (self.buttons_width * 0.3), rect.center[1]) meal_image = pygame_gui.elements.UIImage( relative_rect=rect, image_surface=meal_surface, manager=self.manager, container=container, anchors={"center": "center"}, ) rect = pygame.Rect( (0, 0), (container_width / 4, row_height), ) rect.left = 0 meal_label = pygame_gui.elements.UILabel( text="translations.was_served", relative_rect=rect, manager=self.manager, container=container, object_id="#was_served", anchors={"centery": "centery", "left_target": meal_image}, ) # cook_surface = pygame.Surface( # (row_height, row_height), flags=pygame.SRCALPHA # ) # player_idx = int(player) # player_color = colors[self.vis.player_colors[player_idx]] # self.vis.draw_cook( # screen=cook_surface, # grid_size=row_height, # pos=np.array([row_height / 2, row_height / 2]), # color=player_color, # facing=np.array([0, 1]), # ) # rect = cook_surface.get_rect() # rect.right = 0 # cook_image = pygame_gui.elements.UIImage( # relative_rect=rect, # image_surface=cook_surface, # manager=self.manager, # container=container, # anchors={ # "centery": "centery", # "right": "right", # "right_target": meal_label, # }, # ) last_completed_meals.append(container) self.scroll_space_completed_meals.set_scrollable_area_dimensions( (self.scroll_width_completed_meals * 0.95, container_height) ) def exit_game(self): """Exits the game.""" self.menu_state = MenuStates.PostGame if self.CONNECT_WITH_STUDY_SERVER: self.send_level_done() self.disconnect_websockets() self.update_post_game_screen(self.last_state) self.update_screen_elements() self.beeped_once = False def draw_game(self, state: dict): """Main visualization function. Args: state: The game state returned by the environment. """ self.game_screen = self.vis.draw_gamescreen( state, [int(k.current_player) for k in self.key_sets], ) self.vis.draw_orders( screen=self.orders_image.image, state=state, grid_size=self.screen_margin * 0.68, width=self.orders_container_width, height=self.screen_margin, config=self.visualization_config, ) border = self.visualization_config["GameWindow"]["game_border_size"] border_rect = pygame.Rect( self.window_width // 2 - (self.game_width // 2) - border, self.window_height // 2 - (self.game_height // 2) - border, self.game_width + 2 * border, self.game_height + 2 * border, ) pygame.draw.rect( self.main_window, colors[self.visualization_config["GameWindow"]["game_border_color"]], border_rect, width=border, ) self.update_score_label(state) self.update_remaining_time(state["remaining_time"]) if state["info_msg"]: for idx, msg in enumerate(reversed(state["info_msg"])): text_surface = self.comic_sans.render( msg[0], antialias=True, color=(0, 0, 0) if msg[1] == "Normal" else ((255, 0, 0) if msg[1] == "Warning" else (0, 255, 0)), # bgcolor=(255, 255, 255), ) self.main_window.blit( text_surface, ( self.window_width / 4, self.window_height - self.screen_margin + 5 + (20 * idx), ), ) def draw_game_screen_frame(self, tutorial: bool = False): """Main visualization function for the game screen. Args: tutorial: If True, the tutorial screen is drawn, which is a simplified version of the game screen. """ self.last_state = self.request_state() self.handle_keys() self.handle_joy_stick_input(joysticks=self.joysticks) if tutorial: self.game_screen = self.vis.draw_gamescreen( self.last_state, [int(k.current_player) for k in self.key_sets], ) game_screen_rect = self.game_screen.get_rect() game_screen_rect.center = self.game_center self.main_window.blit(self.game_screen, game_screen_rect) return else: if not self.beeped_once and self.last_state["all_players_ready"]: self.beeped_once = True self.play_bell_sound() if self.last_state["ended"]: self.exit_game() else: self.draw_game(self.last_state) game_screen_rect = self.game_screen.get_rect() game_screen_rect.center = [ self.window_width // 2, self.window_height // 2, ] self.main_window.blit(self.game_screen, game_screen_rect) if not self.last_state["all_players_ready"]: self.wait_players_label.show() else: self.wait_players_label.hide() def update_score_label(self, state: dict): """Updates the score label. Args: state: The game state returned by the environment. """ score = state["score"] self.score_label.set_text( "translations.score", text_kwargs={"score": str(score)} ) if self.switch_score_color: self.count_frames_score_label += 1 duration_color_change = 80 if score > self.last_score: self.score_label.update_theming( '{"colours": {"normal_text": "#03b706"}, "font": { "size": 28, "bold": 1}}' ) self.count_frames_score_label = 0 self.switch_score_color = True elif score < self.last_score: self.score_label.update_theming( '{"colours": {"normal_text": "#e22312"}, "font": { "size": 28, "bold": 1}}' ) self.count_frames_score_label = 0 self.switch_score_color = True elif self.switch_score_color: if self.count_frames_score_label >= duration_color_change: self.score_label.update_theming( '{"colours": {"normal_text": "#000000"}, "font": { "size": 28, "bold": 1}}' ) self.count_frames_score_label = 0 self.switch_score_color = False self.last_score = score def update_remaining_time(self, remaining_time: float): """Updates the remaining time label. Args: remaining_time: The remaining time in seconds. """ hours, rem = divmod(int(remaining_time), 3600) minutes, seconds = divmod(rem, 60) display_time = f"{minutes}:{'%02d' % seconds}" self.timer_label.set_text( "translations.time_remaining", text_kwargs={"time": display_time} ) def create_env_on_game_server(self, tutorial: bool): """Starts an environment on the game server. Args: tutorial: If True, a tutorial environment is created. Else a normal game environment is created. """ if tutorial: layout_path = ROOT_DIR / "configs" / "layouts" / "tutorial.layout" environment_config_path = ROOT_DIR / "configs" / "tutorial_env_config.yaml" else: environment_config_path = ROOT_DIR / "configs" / "environment_config.yaml" layout_path = self.layout_file_paths_dict[self.selected_layout] # layout_path = self.layout_file_paths[self.current_layout_idx] item_info_path = ROOT_DIR / "configs" / "item_info.yaml" item_info, layout, environment_config, _ = load_config_files(item_info_path, layout_path, environment_config_path) num_players = 1 if tutorial else self.number_players seed = int(random.random() * 100000) creation_json = CreateEnvironmentConfig( manager_id=self.manager_id, number_players=num_players, environment_settings={"all_player_can_pause_game": False}, item_info_config=item_info, environment_config=environment_config, layout_config=layout, seed=seed, env_name=layout_path.stem, ).model_dump(mode="json") # print(CreateEnvironmentConfig.model_validate_json(json_data=creation_json)) env_info = requests.post( f"{self.request_url}/manage/create_env/", json=creation_json, ) if env_info.status_code == 403: raise ValueError(f"Forbidden Request: {env_info.json()['detail']}") elif env_info.status_code == 409: log.warning("CONFLICT") env_info = env_info.json() assert isinstance(env_info, dict), "Env info must be a dictionary" self.current_env_id = env_info["env_id"] self.player_info = env_info["player_info"] if tutorial: self.player_ids = [str(list(self.player_info.keys())[0])] if tutorial: self.key_sets = self.setup_player_keys( players=["0"], number_key_sets=1, disjunct=False, ) else: num_key_set = 2 if self.multiple_keysets else 1 self.key_sets = self.setup_player_keys( players=[str(i) for i in range(self.number_humans_to_be_added)], number_key_sets=min(self.number_humans_to_be_added, num_key_set), disjunct=self.split_players, ) self.level_info = env_info self.level_info["name"] = self.selected_layout.replace(".layout", "") self.level_info["number_players"] = num_players def update_pregame_screen(self): """Updates the pregame screen. Possible recipes in the level are displayed.""" self.level_name_label.set_text( "translations.level_name", text_kwargs={"level": self.level_info["name"]} ) rows = 0 for rg in self.level_info["recipe_graphs"]: rows += len(np.unique(np.array(list(rg["layout"].values()))[:, 1])) row_height = self.window_height / 14 container_width = self.window_width * 0.98 container_height = rows * row_height icon_size = row_height * 0.9 graph_width = container_width * 0.55 main_container = pygame_gui.elements.UIPanel( relative_rect=pygame.Rect( (0, 0), ( container_width, container_height, ), ), object_id="#graph_container", manager=self.manager, container=self.scroll_space_recipes, ) last_recipes_labels = [] for idx, rg in enumerate(self.level_info["recipe_graphs"]): meal = rg["meal"] meal = re.sub(r"(?<!^)(?=[A-Z])", " ", meal) positions = np.array(list(rg["layout"].values())) height = row_height * len(np.unique(positions[:, 1])) graph_height = height * 0.9 graph_surface = pygame.Surface( (graph_width, graph_height), flags=pygame.SRCALPHA ) self.vis.draw_recipe_image( graph_surface, rg, graph_width, graph_height, icon_size ) if idx == 0: anchors = {"centerx": "centerx", "top": "top"} else: anchors = { "centerx": "centerx", "top_target": last_recipes_labels[idx - 1], } container = pygame_gui.elements.UIPanel( relative_rect=pygame.Rect( (0, 0), ( container_width, height, ), ), object_id="#graph_container", manager=self.manager, container=main_container, anchors=anchors, ) rect = pygame.Rect( (0, 0), (self.window_width / 6, height), ) label = pygame_gui.elements.UILabel( text=f"translations.{meal}", relative_rect=rect, manager=self.manager, container=container, object_id="#recipe_name", anchors={"centery": "centery", "left": "left"}, ) rect = graph_surface.get_rect() graph_image = pygame_gui.elements.UIImage( relative_rect=rect, image_surface=graph_surface, manager=self.manager, object_id="#recipe_graph", container=container, anchors={"center": "center"}, ) last_recipes_labels.append(container) rect = pygame.Rect( (0, 0), (self.window_width / 6, height), ) rect.right = 0 recipe_score_label = pygame_gui.elements.UILabel( text="", relative_rect=rect, manager=self.manager, container=container, object_id="#recipe_name", anchors={"centery": "centery", "right": "right"}, ) recipe_score_label.set_text( "translations.recipe_score", text_kwargs={"score": str(rg["score"])} ) self.scroll_space_recipes.set_scrollable_area_dimensions( (self.window_width * 0.95, container_height) ) def setup_tutorial(self): """Sets up the tutorial environment. This includes creating the environment on the game server, and setting up the connection.""" answer = requests.post( f"{self.request_url}/connect_to_tutorial/{self.participant_id}" ) if answer.status_code == 200: answer = requests.post( f"{self.request_url}/get_game_connection/{self.participant_id}" ) if answer.status_code == 200: answer_json = answer.json() self.player_info = answer_json["player_info"]["0"] self.level_info = answer_json["level_info"] self.player_info = {self.player_info["player_id"]: self.player_info} else: log.warning("Could not connect to tutorial.") else: self.menu_state = MenuStates.Start log.warning("Could not create tutorial.") def get_game_connection(self): """Sets up a connection to the game server. This includes getting the player info, level info and player keys""" if self.menu_state == MenuStates.Tutorial: self.setup_tutorial() self.key_sets = self.setup_player_keys(["0"], 1, False) self.vis.create_player_colors(1) else: answer = requests.post( f"{self.request_url}/get_game_connection/{self.participant_id}" ) if answer.status_code == 200: answer_json = answer.json() self.player_info = answer_json["player_info"] self.level_info = answer_json["level_info"] self.last_level = self.level_info["last_level"] else: log.warning("COULD NOT GET GAME CONNECTION") self.menu_state = MenuStates.Start self.number_players = -1 if self.split_players: assert ( self.number_humans_to_be_added > 1 ), "Not enough players for key configuration." num_key_set = 2 if self.multiple_keysets else 1 self.key_sets = self.setup_player_keys( list(self.player_info.keys()), min(self.number_humans_to_be_added, num_key_set), self.split_players, ) self.player_ids = list(self.player_info.keys()) def create_and_connect_bot(self, player_id: str, player_info: PlayerInfo): """Creates a bot process and connects it to the game server. Args: player_id: The id/name of the player. player_info: Player info containing client_id, player_hash player_id and websocket_url. """ player_hash = player_info["player_hash"] print( f'--general_plus="agent_websocket:{player_info["websocket_url"]};' f'player_hash:{player_hash};agent_id:{player_id}"' ) if self.USE_AAAMBOS_AGENT: sub = Popen( " ".join( [ "exec", "aaambos", "run", "--arch_config", str(ROOT_DIR / "configs" / "agents" / "arch_config.yml"), "--run_config", str(ROOT_DIR / "configs" / "agents" / "run_config.yml"), f'--general_plus="agent_websocket:{player_info["websocket_url"]};' f'player_hash:{player_hash};agent_id:{player_id}"', f"--instance={player_hash}", ] ), shell=True, ) else: sub = Popen( " ".join( [ "python", str(ROOT_DIR / "configs" / "agents" / "random_agent.py"), f'--uri {player_info["websocket_url"]}', f"--player_hash {player_hash}", f"--player_id {player_id}", f"--recipe_graph '{json.dumps(self.level_info['recipe_graphs'])}'", ] ), shell=True, ) self.sub_processes.append(sub) def connect_websockets(self): """Connects the websockets of the players to the game server. If the player is a bot, a bot process is created""" for p, (player_id, player_info) in enumerate(self.player_info.items()): if p < self.number_humans_to_be_added: # add player websockets websocket = connect(player_info["websocket_url"]) message_dict = { "type": PlayerRequestType.READY.value, "player_hash": player_info["player_hash"], } ws_message = WebsocketMessage(**message_dict) websocket.send(ws_message.json()) assert ( json.loads(websocket.recv())["status"] == 200 ), "not accepted player" self.websockets[player_id] = websocket else: self.create_and_connect_bot(player_id, player_info) if p == 0: self.state_player_id = player_id def setup_game(self): """Sets up prerequisites for the game. This includes connecting the websockets, creating player colors in the vis and setting the kitchen size.""" self.connect_websockets() self.vis.create_player_colors(self.level_info["number_players"]) self.kitchen_width, self.kitchen_height = self.level_info["kitchen_size"] def stop_game_on_server(self, reason: str) -> None: """Stops the game directly on the game server. Args: reason: The reason for stopping the game. """ log.debug(f"Stopping game: {reason}") if not self.CONNECT_WITH_STUDY_SERVER: answer = requests.post( f"{self.request_url}/manage/stop_env/", json={ "manager_id": self.manager_id, "env_id": self.current_env_id, "reason": reason, }, ) if answer.status_code != 200: log.warning( "Could not send level done.", answer.status_code, answer.json()["detail"], ) def send_tutorial_finished(self): """Signals the study server that the tutorial was finished.""" requests.post( f"{self.request_url}/disconnect_from_tutorial/{self.participant_id}", ) def finished_button_press(self): """Gets called when the finished button is pressed. Stops the game on the game server if the study server is not used.""" if not self.CONNECT_WITH_STUDY_SERVER: self.stop_game_on_server("finished_button_pressed") self.menu_state = MenuStates.PostGame log.debug("Pressed finished button") self.update_screen_elements() def fullscreen_button_press(self): """Toggles between fullscreen and windowed mode.""" self.fullscreen = not self.fullscreen self.set_window_size() self.init_ui_elements() self.set_game_size() self.update_screen_elements() # self.setup_layout_selection() def reset_gui_values(self): """Reset the values of the GUI elements to their default values. Default values are defined here.""" self.currently_controlled_player_idx = 0 self.number_humans_to_be_added = 1 self.number_bots_to_be_added = 0 self.number_players = ( self.number_humans_to_be_added + self.number_bots_to_be_added ) self.split_players = False self.multiple_keysets = False def update_selection_elements(self): """Updates the selection elements of the GUI. This includes the number of players, the number of bots, the split players button and the multiple keysets button.""" if self.number_humans_to_be_added == 1: self.remove_human_button.disable() self.multiple_keysets_button.hide() self.split_players_button.hide() else: self.remove_human_button.enable() if self.number_humans_to_be_added < 2: self.multiple_keysets_button.hide() else: self.multiple_keysets_button.show() if self.number_humans_to_be_added > 1 and self.multiple_keysets: self.split_players_button.show() else: self.split_players_button.hide() if self.number_bots_to_be_added == 0: self.remove_bot_button.disable() else: self.remove_bot_button.enable() self.number_players = ( self.number_humans_to_be_added + self.number_bots_to_be_added ) text = ( "translations.2_players" if self.multiple_keysets else "translations.1_player" ) self.multiple_keysets_button.set_text(text) self.added_players_label.set_text( "translations.humans_to_be_added", text_kwargs={"number": str(self.number_humans_to_be_added)}, ) self.added_bots_label.set_text( "translations.bots_to_be_added", text_kwargs={"number": str(self.number_bots_to_be_added)}, ) text = ( "translations.split_players_yes" if self.split_players else "translations.split_players_no" ) self.split_players_button.set_text(text) if self.number_players == 0: self.start_button.disable() else: self.start_button.enable() self.selected_layout_path = self.layout_file_paths_dict[self.selected_layout] self.selected_layout_label.set_text( "translations.selected_layout", text_kwargs={"layout": self.selected_layout.replace(".layout", "")} ) if self.layout_surfaces: print(self.selected_layout_label.rect) # between label and start button max_target_height = ((self.window_height // 2) - self.buttons_height - (self.buttons_height // 2)) * 0.8 new_surf = pygame.transform.scale_by(self.layout_surfaces[self.selected_layout], max_target_height / self.layout_surfaces[ self.selected_layout].get_height()) self.preview_layout_image.set_dimensions(new_surf.get_size()) self.preview_layout_image.set_image(new_surf) def send_action(self, action: Action): """Sends an action to the game environment. Args: action: The action to be sent. Contains the player, action type and move direction if action is a movement. """ if isinstance(action.action_data, np.ndarray): action.action_data = [ float(action.action_data[0]), float(action.action_data[1]), ] message_dict = { "type": PlayerRequestType.ACTION.value, "action": action, "player_hash": self.player_info[action.player]["player_hash"], } ws_message = WebsocketMessage(**message_dict) self.websockets[action.player].send(ws_message.json()) self.websockets[action.player].recv() def request_state(self): """Requests the current state of the game environment from the game server.""" message_dict = { "type": PlayerRequestType.GET_STATE.value, "action": None, "player_hash": self.player_info[self.state_player_id]["player_hash"], } ws_message = WebsocketMessage(**message_dict) self.websockets[self.state_player_id].send(ws_message.json()) state = json.loads(self.websockets[self.state_player_id].recv()) return state def exit_tutorial(self): """Exits the tutorial. Disconnects the websockets and signals the study server that the tutorial was finished if the study server is used. Otherwise, the game is stopped on the game server. """ self.disconnect_websockets() self.menu_state = MenuStates.PreGame if self.CONNECT_WITH_STUDY_SERVER: self.send_tutorial_finished() else: self.stop_game_on_server("tutorial_finished") def disconnect_websockets(self): """Disconnects the websockets. Kills all subprocesses that are running bots.""" for sub in self.sub_processes: try: if self.USE_AAAMBOS_AGENT: pgrp = os.getpgid(sub.pid) os.killpg(pgrp, signal.SIGINT) subprocess.run( "kill $(ps aux | grep 'aaambos' | awk '{print $2}')", shell=True ) else: sub.kill() except ProcessLookupError: pass self.sub_processes = [] for websocket in self.websockets.values(): websocket.close() @staticmethod def play_bell_sound(): """Plays a bell sound when the game starts.""" bell_path = str(ROOT_DIR / "pygame_2d_vis" / "sync_bell.wav") mixer.init() mixer.music.load(bell_path) mixer.music.set_volume(0.7) mixer.music.play() log.log(logging.INFO, "Started game, played bell sound") def start_study(self): """Starts the study on the study server.""" answer = requests.post( f"{self.request_url}/start_study/{self.participant_id}/{self.number_humans_to_be_added}" ) if answer.status_code == 200: self.last_level = False self.get_game_connection() else: self.menu_state = MenuStates.Start log.warning( "COULD NOT START STUDY; Response:", answer.status_code, answer.json()["detail"], ) def send_level_done(self): """Sends a message to the study server that the level was finished.""" answer = requests.post(f"{self.request_url}/level_done/{self.participant_id}") if answer.status_code != 200: log.warning( "Could not send level done.", answer.status_code, answer.json()["detail"], ) def button_continue_postgame_pressed(self): """Handles the continue button press on the postgame screen. If the study server is used, the connection to a new game is set up. Otherwise, the start screen is shown. """ if self.CONNECT_WITH_STUDY_SERVER: if not self.last_level: self.get_game_connection() else: self.menu_state = MenuStates.Start return self.menu_state = MenuStates.PreGame def manage_button_event(self, button: pygame_gui.core.UIElement | None = None): """Manages the button events. The button events are filtered by the current menu state and the button that was pressed. Args: button: The button that was pressed. """ if button == self.quit_button: if self.fullscreen: self.fullscreen_button_press() self.running = False self.disconnect_websockets() self.stop_game_on_server("Quit button") self.menu_state = MenuStates.Start log.debug("Pressed quit button") return elif button == self.fullscreen_button: self.fullscreen_button_press() log.debug("Pressed fullscreen button") return # Filter by shown screen page match self.menu_state: ############################################ case MenuStates.Start: match button: case self.start_button: if not ( self.number_humans_to_be_added + self.number_bots_to_be_added ): pass else: self.menu_state = MenuStates.Tutorial if self.CONNECT_WITH_STUDY_SERVER: self.get_game_connection() else: self.create_env_on_game_server(tutorial=True) self.setup_game() case self.add_human_player_button: self.number_humans_to_be_added += 1 case self.add_bot_button: self.number_bots_to_be_added += 1 case self.remove_human_button: self.number_humans_to_be_added = max( 0, self.number_humans_to_be_added - 1 ) case self.remove_bot_button: self.number_bots_to_be_added = max( 0, self.number_bots_to_be_added - 1 ) case self.multiple_keysets_button: self.multiple_keysets = not self.multiple_keysets self.split_players = False case self.split_players_button: self.split_players = not self.split_players case other: if button in self.layout_buttons: self.selected_layout = self.layout_buttons[button] ############################################ case MenuStates.Tutorial: self.exit_tutorial() if self.CONNECT_WITH_STUDY_SERVER: self.start_study() else: self.create_env_on_game_server(tutorial=False) ############################################ case MenuStates.PreGame: if button in [self.continue_button, self.continue_button_two]: self.setup_game() self.set_game_size() self.menu_state = MenuStates.Game ############################################ case MenuStates.Game: if button in [self.return_to_main_button]: if not self.CONNECT_WITH_STUDY_SERVER: self.menu_state = MenuStates.Start self.stop_game_on_server("Return to main button") self.disconnect_websockets() ############################################ case MenuStates.PostGame: if button in [ self.finish_study_button, self.next_game_button, self.continue_button, self.continue_button_two ]: if self.last_level: self.menu_state = MenuStates.End else: self.button_continue_postgame_pressed() ############################################ def add_joystick(self, event: pygame.event.Event): """Adds a joystick to the list of joysticks and assigns it to a key set. A pygame.JOYDEVICEADDED event is generated for every joystick connected at the start of the program. Args: event: The event that is triggered when a joystick is connected. """ joy = pygame.joystick.Joystick(event.device_index) self.joysticks[joy.get_instance_id()] = joy for key_set in self.key_sets: if key_set.joystick is None: key_set.joystick = joy.get_instance_id() break log.debug(f"Joystick {joy.get_instance_id()} connected") def remove_joystick(self, event: pygame.event.Event): """Removes a joystick from the list of joysticks and unassigns it from a key set. Args: event: The event that is triggered when a joystick is disconnected. """ del self.joysticks[event.instance_id] for key_set in self.key_sets: if key_set.joystick == event.instance_id: key_set.joystick = None log.debug(f"Joystick {event.instance_id} disconnected") log.debug(f"Number of joysticks:" + str(pygame.joystick.get_count())) def process_gui_event(self, event: pygame.event.Event): """Processes the pygame events. The events are filtered by their type and the current menu state. Args: event: The pygame event that is processed. """ if event.type == pygame.QUIT: self.disconnect_websockets() self.running = False if event.type == pygame.JOYDEVICEADDED and pygame.joystick.get_count() > 0: self.add_joystick(event) if event.type == pygame.JOYDEVICEREMOVED: self.remove_joystick(event) if self.visualization_config["Gui"]["press_button_to_continue"] and ( event.type == pygame.JOYBUTTONDOWN and any([joy.get_button(7) for joy in self.joysticks.values()]) or (event.type == pygame.KEYDOWN and event.key == pygame.K_RETURN) ): if self.menu_state == MenuStates.Start: self.manage_button_event(self.start_button) self.update_screen_elements() elif self.menu_state in [ MenuStates.Tutorial, MenuStates.PreGame, MenuStates.PostGame, ]: self.manage_button_event(self.continue_button) self.update_screen_elements() if event.type == pygame_gui.UI_BUTTON_PRESSED: button = event.ui_element self.manage_button_event(button) if button in [ self.start_button, self.continue_button, self.continue_button_two, self.finish_study_button, self.next_game_button, self.return_to_main_button ]: self.update_screen_elements() elif self.menu_state == MenuStates.Start: self.update_selection_elements() if self.menu_state in [MenuStates.Game, MenuStates.Tutorial]: if event.type in [pygame.KEYDOWN, pygame.KEYUP]: self.handle_key_event(event) if event.type in [pygame.JOYBUTTONDOWN, pygame.JOYBUTTONUP]: self.handle_joy_stick_event(event, joysticks=self.joysticks) self.manager.process_events(event) def start_pygame(self): """Starts pygame and the gui loop. Each frame the game state is visualized and keyboard inputs are read.""" log.debug(f"Starting pygame gui at {self.FPS} fps") pygame.font.init() self.comic_sans = pygame.font.SysFont("Comic Sans MS", 30) pygame.display.set_caption("Cooperative Cuisine") clock = pygame.time.Clock() self.set_window_size() self.init_ui_elements() self.reset_gui_values() self.update_screen_elements() # Game loop self.running = True # pygame.JOYDEVICEADDED event is generated for every joystick connected at the start of the program. self.joysticks = {} while self.running: try: self.time_delta = clock.tick(self.FPS) / 1000 # PROCESSING EVENTS for event in pygame.event.get(): self.process_gui_event(event) # DRAWING self.draw_main_window() except (KeyboardInterrupt, SystemExit): self.running = False self.disconnect_websockets() if not self.CONNECT_WITH_STUDY_SERVER: self.stop_game_on_server("Program exited.") if self.fullscreen: self.fullscreen_button_press() pygame.quit() sys.exit() def main( study_url: str, study_port: int, game_url: str, game_port: int, manager_ids: list[str], USE_AAAMBOS_AGENT=False, debug=False, do_study=False, use_ssl=False, ): setup_logging() gui = PyGameGUI( study_host=study_url, study_port=study_port, game_host=game_url, game_port=game_port, manager_ids=manager_ids, USE_AAAMBOS_AGENT=USE_AAAMBOS_AGENT, debug=debug, do_study=do_study, use_ssl=use_ssl, ) gui.start_pygame() if __name__ == "__main__": parser = argparse.ArgumentParser( prog="Cooperative Cuisine 2D PyGame Visualization", description="PyGameGUI: a PyGame 2D Visualization window.", epilog="For further information, " "see https://scs.pages.ub.uni-bielefeld.de/cocosy/cooperative-cuisine/overcooked_simulator.html", ) create_gui_parser(parser) args = parser.parse_args() main( args.study_url, args.study_port, args.game_url, args.game_port, args.manager_ids, debug=args.debug, do_study=args.do_study, use_ssl=args.ssl, )