Branch

Sybe/TODO.txt

@@ -0,0 +1,26 @@
+
+
+
+### DONE! ###
+- Fixed Python version conflict
+- Repaired pip, created virtual environment
+- Fixed dependency hell with mismatching versions of libraries
+
+- Figured out how to generate game data with 2 random agents
+- Implemented manual stitching of pngs to create a gif
+- Created an interactive viewer in a pygame environment
+- Added HUD to viewer, improved UX with tap/hold of buttons.
+- Restructured code; created file hierarchy
+- Created next run function and improved pathing
+
+### Issues: ###
+- (base) env? check correctness
+- 
+
+### TODO: ###
+- 
+- Replace random agents with reinforcement learning agents
+- Train on the JaxMARL server
+- Save trajectory instead of gif
+
+

Sybe/archive/make_gif.py

@@ -0,0 +1,189 @@
+import os
+import time
+import pygame
+import imageio
+from overcooked_ai_py.agents.agent import AgentPair, RandomAgent
+from overcooked_ai_py.agents.benchmarking import AgentEvaluator
+from overcooked_ai_py.visualization.state_visualizer import StateVisualizer
+
+# --- Configuration ---
+LAYOUT_NAME = "cramped_room"
+HORIZON = 400
+FPS = 30
+OUTPUT_DIR = os.path.join("Sybe", "experiments", "game1")
+
+# --- Utility Functions ---
+
+def generate_game_data(layout_name, horizon):
+    """Runs a single simulation with random agents and returns the history data."""
+    print(f"Simulating game on '{layout_name}' for {horizon} steps...")
+
+    # 1. Setup Evaluator
+    agent_evaluator = AgentEvaluator.from_layout_name(
+        mdp_params={"layout_name": layout_name},
+        env_params={"horizon": horizon}
+    )
+
+    # 2. Run Game
+    agent_pair = AgentPair(RandomAgent(), RandomAgent())
+    # The full history is a dict of lists containing states, actions, rewards, etc.
+    game_history = agent_evaluator.evaluate_agent_pair(agent_pair, num_games=1)
+
+    # 3. Extract Core Data (assuming only one game was run)
+    state_list = game_history["ep_states"][0]
+    grid_matrix = agent_evaluator.env.mdp.terrain_mtx
+
+    return game_history, state_list, grid_matrix, agent_evaluator
+
+def stitch_gif(game_history, output_dir, gif_name="interactive_replay.gif", gif_fps=10):
+    """Saves individual frames to a directory and stitches them into a GIF."""
+
+    if not os.path.exists(output_dir):
+        os.makedirs(output_dir)
+
+    print(f"Rendering frames to {output_dir}...")
+
+    # 1. Render all frames to PNG files
+    StateVisualizer().display_rendered_trajectory(
+        trajectories=game_history,
+        img_directory_path=output_dir,
+        ipython_display=False
+    )
+
+    # 2. Stitch images into a GIF
+    print("Stitching frames into GIF...")
+    images = []
+
+    # Get all PNG filenames and sort them numerically
+    filenames = sorted(
+        [f for f in os.listdir(output_dir) if f.endswith('.png')],
+        key=lambda x: int(''.join(filter(str.isdigit, x)))
+    )
+
+    for filename in filenames:
+        file_path = os.path.join(output_dir, filename)
+        images.append(imageio.imread(file_path))
+
+    output_path = os.path.join(output_dir, gif_name)
+    imageio.mimsave(output_path, images, duration=(1/gif_fps), loop=0)
+
+    print(f"GIF successfully saved to: {output_path}")
+
+def run_interactive_viewer(game_history, state_list, grid_matrix, mdp_env, screen_fps=30):
+    """Opens a Pygame window to interactively replay the game, including HUD."""
+
+    pygame.init()
+
+    # --- KEY REPEAT SETUP ---
+    # Argument 1: Delay before repeat starts (500ms = 0.5 seconds)
+    # Argument 2: Interval between repeats (30ms = ~33 steps per second)
+    pygame.key.set_repeat(500, 30) 
+
+    viz = StateVisualizer()
+    font = pygame.font.SysFont(None, 24) 
+
+    rewards_list = game_history["ep_rewards"][0] 
+
+    def get_game_surface(t):
+        return viz.render_state(state_list[t], grid_matrix)
+
+    def draw_hud(surface, timestep):
+        total_score = int(game_history["ep_returns"][0])
+
+        if timestep == 0:
+            current_score = 0
+        else:
+            current_score = int(sum(rewards_list[:timestep]))
+
+        max_time = mdp_env.horizon
+        time_left = max(0, max_time - timestep)
+
+        # Text Rendering
+        text_time = font.render(f"TIME LEFT: {time_left}", True, (255, 255, 255))
+        text_score = font.render(f"SCORE: {current_score} / {total_score}", True, (255, 255, 255))
+
+        surface.blit(text_time, (20, 10))
+        surface.blit(text_score, (20, 35))
+
+    # Setup Window
+    game_surface = get_game_surface(0)
+    game_width, game_height = game_surface.get_size()
+
+    HUD_HEIGHT = 60
+    screen = pygame.display.set_mode((game_width, game_height + HUD_HEIGHT))
+    clock = pygame.time.Clock()
+
+    # State variables
+    running = True
+    current_timestep = 0
+    playing = False
+
+    BG_COLOR = (153, 102, 0) 
+
+    print("\n--- INTERACTIVE CONTROLS ---")
+    print("Tap/Hold Right : Step Forward")
+    print("Tap/Hold Left  : Step Backward")
+    print("Space          : Play / Pause")
+    print("Esc            : Quit")
+    print("----------------------------")
+
+    while running:
+        # 1. Event Handling (Handles Taps AND Holds via set_repeat)
+        for event in pygame.event.get():
+            if event.type == pygame.QUIT:
+                running = False
+            elif event.type == pygame.KEYDOWN:
+                if event.key == pygame.K_ESCAPE:
+                    running = False
+                elif event.key == pygame.K_RIGHT:
+                    current_timestep = min(current_timestep + 1, len(state_list) - 1)
+                    playing = False # Stop auto-play on manual input
+                elif event.key == pygame.K_LEFT:
+                    current_timestep = max(current_timestep - 1, 0)
+                    playing = False
+                elif event.key == pygame.K_SPACE:
+                    playing = not playing
+
+        # 2. Auto-Play Logic
+        if playing:
+            if current_timestep < len(state_list) - 1:
+                current_timestep += 1
+                # We use clock.tick to control speed, so no sleep needed here
+            else:
+                playing = False
+
+        # 3. Drawing
+        screen.fill(BG_COLOR) 
+
+        draw_hud(screen, current_timestep)
+
+        game_surface = get_game_surface(current_timestep)
+        screen.blit(game_surface, (0, HUD_HEIGHT)) 
+
+        pygame.display.set_caption(f"Replay - Step {current_timestep}/{len(state_list)-1} {'[PLAYING]' if playing else '[PAUSED]'}")
+        pygame.display.flip()
+
+        clock.tick(screen_fps)
+
+    pygame.quit()
+
+
+# --- Main Execution Block ---
+if __name__ == "__main__":
+
+    # 1. Run Simulation
+    game_history, state_list, grid_matrix, agent_evaluator = generate_game_data(
+        layout_name=LAYOUT_NAME, 
+        horizon=HORIZON
+    )
+
+    # 2. Save GIF (Uncomment this line to generate the GIF)
+    # stitch_gif(game_history, output_dir=OUTPUT_DIR, gif_fps=10)
+
+    # 3. Open Interactive Viewer 
+    run_interactive_viewer(
+        game_history=game_history, 
+        state_list=state_list, 
+        grid_matrix=grid_matrix,
+        mdp_env=agent_evaluator.env 
+    )

Sybe/archive/visualise.py

@@ -0,0 +1,13 @@
+from overcooked_ai_py.mdp.overcooked_mdp import OvercookedGridworld
+from overcooked_ai_py.mdp.overcooked_env import OvercookedEnv
+
+# Load a classic layout (e.g., "cramped_room")
+# Other layouts include: "asymmetric_advantages", "coordination_ring", "forced_coordination", "counter_circuit_o_1order"
+mdp = OvercookedGridworld.from_layout_name("cramped_room")
+env = OvercookedEnv.from_mdp(mdp, horizon=400)
+
+# Reset the environment
+env.reset()
+
+# Print the text-based representation of the grid
+print(env)

Sybe/archive/viz_as_png.py

@@ -0,0 +1,46 @@
+import os
+from overcooked_ai_py.mdp.overcooked_mdp import OvercookedGridworld
+from overcooked_ai_py.mdp.overcooked_env import OvercookedEnv
+from overcooked_ai_py.visualization.state_visualizer import StateVisualizer
+from overcooked_ai_py.mdp.actions import Action
+import random
+
+# 1. Setup the environment
+layout_name = "cramped_room"
+mdp = OvercookedGridworld.from_layout_name(layout_name)
+env = OvercookedEnv.from_mdp(mdp, horizon=400)
+env.reset()
+
+# 2. Get a state (just stepping randomly for demonstration)
+# Taking a random action for agent 0 and agent 1
+action_0 = random.choice(Action.ALL_ACTIONS)
+action_1 = random.choice(Action.ALL_ACTIONS)
+state, reward, done, info = env.step((action_0, action_1))
+
+print("actions taken:", action_0, action_1)
+print("Current State after one random step:")
+print(state)
+print("Reward:", reward)
+print("Done:", done)
+print("Info:", info)
+
+# 3. Create the visualizer
+visualizer = StateVisualizer()
+
+# 4. Render and Save
+# You need the 'grid' from the mdp to tell the visualizer where the walls/counters are
+img_dir = "Sybe/replays"
+if not os.path.exists(img_dir):
+    os.makedirs(img_dir)
+
+img_path = os.path.join(img_dir, "test_state.png")
+
+print(f"Saving image to {img_path}...")
+visualizer.display_rendered_state(
+    state=state, 
+    grid=mdp.terrain_mtx, 
+    img_path=img_path
+)
+print("Done!")
+
+StateVisualizer().display_rendered_trajectory(trajs2, ipython_display=True)

Sybe/config.py

@@ -0,0 +1,26 @@
+"""Central place for constants and paths."""
+from pathlib import Path
+
+LAYOUT_NAME = "cramped_room"
+HORIZON = 400
+
+VIEWER_FPS = 30
+GIF_FPS = 10
+HUD_HEIGHT = 60
+BG_COLOR = (153, 102, 0)
+
+EXPERIMENTS = [
+    "random",
+    "ddqn_baseline",
+    "independent_curiosity",
+]
+
+# AGENT_PAIRS = {
+#     "random": ("random", "random"),
+#     "ddqn_baseline": ("ddqn", "ddqn"),
+#     "independent_curiosity": ("curiosity", "curiosity"),
+# }
+
+# Paths are relative to this file, so scripts work from any working directory
+BASE_DIR = Path(__file__).resolve().parent
+# OUTPUT_DIR = BASE_DIR / "experiments" / "random" / "run_000001" / "renders" / "episode_000004"

Sybe/main.py

@@ -0,0 +1,58 @@
+"""Entry point: simulate a game.
+
+Run from anywhere:
+    python Sybe/main.py                             # simulate a game
+    python Sybe/main.py --save_frames               # save frames
+    python Sybe/main.py --viewer                    # show pygame viewer
+"""
+import argparse
+
+from overcooked_irl.agent_factory import create_agent_pair
+import config
+from overcooked_irl.export import save_frames, save_gif
+from overcooked_irl.simulation import random_agent_pair, run_game
+from overcooked_irl.viewer import run_viewer
+from overcooked_irl.paths import next_episode_dir, next_run_dir
+from overcooked_ai_py.visualization.state_visualizer import StateVisualizer # TODO Remove StateVisualizer logic from main?
+
+def parse_args(): 
+    p = argparse.ArgumentParser(description="Overcooked simulation + visualization")
+    p.add_argument("--layout", default=config.LAYOUT_NAME)
+    p.add_argument("--horizon", type=int, default=config.HORIZON)
+    p.add_argument(
+        "--experiment",
+        choices=config.EXPERIMENTS,
+        default="random",
+        help="Experiment configuration",
+    )
+    p.add_argument("--save_frames", action="store_true", help="saves images of every timestep")
+    p.add_argument("--save_gif", action="store_true", help="save images AND a gif")
+    p.add_argument("--viewer", action="store_true", help="show the pygame viewer")
+    return p.parse_args()
+
+
+def main():
+    args = parse_args()
+
+    run_dir = config.BASE_DIR / "experiments" / args.experiment / "run_0001" 
+    # run_dir = next_run_dir(args.experiment)
+
+    agent_pair = create_agent_pair(args.experiment, args.experiment)
+
+    episode = run_game(agent_pair, args.layout, args.horizon)
+    episode_dir = run_dir / "episode_000001"
+    # episode_dir = next_episode_dir(run_dir)
+
+    if args.save_gif:
+        frames_dir = episode_dir / "frames"
+        save_gif(episode.trajectories, frames_dir, fps=config.GIF_FPS)
+    elif args.save_frames:
+        save_frames(episode.trajectories, frames_dir)
+
+    if args.viewer:
+        run_viewer(episode)
+
+
+
+if __name__ == "__main__":
+    main()

Sybe/overcooked_irl/agent_factory.py

@@ -0,0 +1,21 @@
+from overcooked_ai_py.agents.agent import AgentPair, RandomAgent
+
+def create_agent(agent_type):
+    if agent_type == "random":
+        return RandomAgent()
+
+    elif agent_type == "ddqn":
+        raise NotImplementedError
+
+    elif agent_type == "curiosity":
+        raise NotImplementedError
+
+    else:
+        raise ValueError(f"Unknown agent type: {agent_type}")
+
+
+def create_agent_pair(agent1_type, agent2_type):
+    agent1 = create_agent(agent1_type)
+    agent2 = create_agent(agent2_type)
+
+    return AgentPair(agent1, agent2)