-
Florian Schröder authoredWith this commit, the game now supports SSL connections. In addition, the argument parsing workflow was fundamentally improved to provide a clearer, better-structured usage of command line arguments for different functionalities. It also includes updates to the help documentations - now, each of these functionalities are accessible as subcommands with their own help messages. A minor update was also made to the test suite to reflect these changes.
Florian Schröder authoredWith this commit, the game now supports SSL connections. In addition, the argument parsing workflow was fundamentally improved to provide a clearer, better-structured usage of command line arguments for different functionalities. It also includes updates to the help documentations - now, each of these functionalities are accessible as subcommands with their own help messages. A minor update was also made to the test suite to reflect these changes.
video_replay.py 12.88 KiB
"""
Generate images from a recording of json states or from the recording of the actions.
The json state files grow very fast. Therefore, we recommend using the actions recording and create a video replay
based on the actions. Until now, we did not find any deviations from the json state visualisation.
# CLI
Sequence of images replay from actions:
```bash
cooperative-cuisine replay -a ~/.local/state/cooperative_cuisine/log/ENV_NAME/actions.jsonl -e ~/.local/state/cooperative_cuisine/log/ENV_NAME/env_configs.jsonl -d -n 2 -p "0"
```
Sequence of images replay from json states:
```bash
cooperative-cuisine replay -j ~/.local/state/cooperative_cuisine/log/ENV_NAME/json_states.jsonl -d -p "0"
```
The `display` (`-d`, `--display`) requires `opencv-python` (cv2) installed. (`pip install opencv-python`)
Generate a video file from images (requires also `opencv-python`):
```bash
cooperative-cuisine replay --video ~/.local/state/cooperative_cuisine/log/ENV_NAME/DIR_NAME_WITH_IMAGES
```
For additional CLI arguments:
```bash
cooperative-cuisine replay -h
```
# Code Documentation
"""
import json
import os
import os.path
from argparse import ArgumentParser
from datetime import datetime, timedelta
from pathlib import Path
import pygame
import yaml
from PIL import Image
from tqdm import tqdm
from cooperative_cuisine.action import Action
from cooperative_cuisine.argument_parser import (
create_replay_parser,
)
from cooperative_cuisine.environment import Environment
from cooperative_cuisine.pygame_2d_vis.drawing import Visualizer
from cooperative_cuisine.recording import FileRecorder
FPS_DEFAULT = 24
STEP_DURATION_DEFAULT = 200
GRID_SIZE_DEFAULT = 40
NUMBER_PLAYERS_DEFAULT = 1
def simulate(
action_file,
env_config,
viz_config,
target_directory,
player_id_filter=None,
step_duration=1 / STEP_DURATION_DEFAULT,
fps=FPS_DEFAULT,
display=False,
number_player=NUMBER_PLAYERS_DEFAULT,
break_when_no_action=False,
grid_size=GRID_SIZE_DEFAULT,
):
"""
Create images by simulating the game environment using a sequence of actions.
You can record the relevant files via hooks in the environment_config:
```yaml
hook_callbacks
env_configs:
func: !!python/name:cooperative_cuisine.hooks.hooks_via_callback_class ''
kwargs:
hooks: [ env_initialized, item_info_config ]
callback_class: !!python/name:cooperative_cuisine.recording.FileRecorder ''
callback_class_kwargs:
record_path: USER_LOG_DIR/ENV_NAME/LOG_RECORD_NAME.jsonl
add_hook_ref: true
actions:
func: !!python/name:cooperative_cuisine.hooks.hooks_via_callback_class ''
kwargs:
hooks: [ pre_perform_action ]
callback_class: !!python/name:cooperative_cuisine.recording.FileRecorder ''
callback_class_kwargs:
record_path: USER_LOG_DIR/ENV_NAME/LOG_RECORD_NAME.jsonl
```
You can call simulation function via the command line. For example by replacing the ENVIRONMENT_ID (Linux system) or the complete path:
```bash
python video_replay.py -a ~/.local/state/cooperative_cuisine/log/ENV_NAME/actions.jsonl -e ~/.local/state/cooperative_cuisine/log/ENV_NAME/env_configs.jsonl -d -n 2 -p "0"
```
Args:
action_file: Path of the file containing the actions in JSON format.
env_config: Path of the environment configuration file.
viz_config: Path of the visualization configuration file.
target_directory: Path of the directory to save the generated images.
player_id_filter: (optional) Filter for player ids.
step_duration: (optional) Duration of each step in seconds. Default is 1 / `STEP_DURATION_DEFAULT`.
fps: (optional) Frames per second for visualization. Default is `FPS_DEFAULT`.
display: (optional) Whether to display the visualization. Default is `False`.
number_player: (optional) Number of players. Default is `NUMBER_PLAYERS_DEFAULT`.
break_when_no_action: (optional) Whether to stop simulation when there are no more actions. Default is `False`.
grid_size: (optional) Size of the grid in pixels. Default is `GRID_SIZE_DEFAULT`.
"""
if display:
import cv2
if player_id_filter is None:
player_id_filter = "0"
if not os.path.isdir(os.path.expanduser(target_directory)):
os.makedirs(os.path.expanduser(target_directory), exist_ok=True)
env_data = {
"item_info_config": None,
"environment_config": None,
"seed": None,
"layout_config": None,
}
with open(os.path.expanduser(env_config), "r") as f:
for line in f:
e = json.loads(line)
# or just du update?
for k in env_data.keys():
if k in e:
env_data[k] = e[k]
env = Environment(
env_config=env_data["environment_config"],
layout_config=env_data["layout_config"],
item_info=env_data["item_info_config"],
seed=env_data["seed"],
as_files=False,
)
env.reset_env_time()
# remove recorder hooks
for key, callback_list in env.hook.hooks.items():
recorder_idx = []
for idx, callback in enumerate(callback_list):
if isinstance(callback, FileRecorder):
recorder_idx.append(idx)
for idx in reversed(recorder_idx):
callback_list.pop(idx)
for p in range(number_player):
env.add_player(f"{p}")
viz = Visualizer(viz_config)
viz.create_player_colors(number_player)
pygame.init()
pygame.font.init()
action_idx = 0
next_action_time = env.env_time
next_frame_time = env.env_time
with open(os.path.expanduser(action_file), "r") as file:
actions = file.readlines()
with tqdm(
total=len(actions)
if break_when_no_action
else int((env.env_time_end - env.env_time).total_seconds())
) as pbar:
while not env.game_ended:
if (
not break_when_no_action
and env.env_time.timestamp() - int(env.env_time.timestamp())
< step_duration
):
pbar.update(1)
env.step(timedelta(seconds=step_duration))
while action_idx < len(actions) and next_action_time <= env.env_time:
action = json.loads(actions[action_idx])
next_action_time = datetime.fromisoformat(action["env_time"])
if next_action_time <= env.env_time:
env.perform_action(Action(**action["action"]))
action_idx += 1
if break_when_no_action:
pbar.update(1)
if break_when_no_action and action_idx >= len(actions):
break
if next_frame_time <= env.env_time:
next_frame_time += timedelta(seconds=1 / fps)
state = env.get_json_state(player_id_filter)
state = json.loads(state)
output_file = Path(os.path.expanduser(target_directory)) / (
state["env_time"] + ".jpg"
)
# viz.save_state_image(grid_size=40, state=state, filename=output_file)
image = viz.get_state_image(grid_size=grid_size, state=state).transpose(
(1, 0, 2)
)
Image.fromarray(image).save(output_file)
if display:
cv2.imshow("Replay", image[:, :, ::-1])
cv2.waitKey(1)
def from_json_states(
json_states_file,
viz_config,
target_directory,
player_id_filter=None,
display=False,
grid_size=GRID_SIZE_DEFAULT,
):
"""
Generate images from recorded json strings in on jsonl file.
For single image creation based on one json state see `cooperative_cuisine.pygame_2d_vis.drawing`.
You can create the jsonl file recording via hooks in the environment_config. These files grow very fast!:
```yaml
json_states:
func: !!python/name:cooperative_cuisine.hooks.hooks_via_callback_class ''
kwargs:
hooks: [ json_state ]
callback_class: !!python/name:cooperative_cuisine.recording.FileRecorder ''
callback_class_kwargs:
record_path: USER_LOG_DIR/ENV_NAME/json_states.jsonl
```
You can call this function via the command line:
```bash
python video_replay.py -j ~/.local/state/cooperative_cuisine/log/ENV_NAME/json_states.jsonl -d -p "0"
```
Args:
json_states_file: The path to the JSON file containing the game states.
viz_config: The visualization configuration.
target_directory: The directory where the generated images will be saved.
player_id_filter: (Optional) The ID of the player to filter by. If None, all players will be included.
display: (Optional) A flag indicating whether to display the generated images using OpenCV.
grid_size: (Optional) The size of the grid to render the game state on.
"""
if display:
import cv2
created_colors = False
if not os.path.isdir(target_directory):
os.makedirs(target_directory, exist_ok=True)
with open(os.path.expanduser(json_states_file), "r") as file:
viz = Visualizer(viz_config)
pygame.init()
pygame.font.init()
for line in tqdm(file):
state = json.loads(line)
if not created_colors:
viz.create_player_colors(len(state["players"]))
created_colors = True
if player_id_filter is None or player_id_filter == state["player_id"]:
output_file = Path(target_directory) / (state["env_time"] + ".jpg")
image = viz.get_state_image(grid_size=grid_size, state=state)
Image.fromarray(image).save(output_file)
if display:
cv2.imshow("Replay", image.transpose((1, 0, 2))[:, :, ::-1])
cv2.waitKey(1)
def video_from_images(image_paths, video_name, fps):
"""Generate a video from images in a directory.
Requires opencv installed.
CLI:
```bash
python video_replay.py --video ~/.local/state/cooperative_cuisine/log/ENV_NAME/DIR_NAME_WITH_IMAGES
```
Args:
image_paths: A string representing the path to the directory containing the images.
video_name: A string representing the name of the output video file.
fps: An integer representing the frames per second for the video.
"""
assert os.path.isdir(image_paths), "image path is not a directory"
import cv2
images = sorted(os.listdir(os.path.expanduser(image_paths)))
frame = cv2.imread(os.path.expanduser(os.path.join(image_paths, images[0])))
height, width, layers = frame.shape
video = cv2.VideoWriter(
os.path.expanduser(video_name),
cv2.VideoWriter_fourcc(*"mp4v"),
fps,
(width, height),
)
for image in tqdm(images):
video.write(cv2.imread(os.path.expanduser(os.path.join(image_paths, image))))
cv2.destroyAllWindows()
print("Generate Video...")
video.release()
print("See:", video_name)
def main(cli_args):
"""
Runs the video replay with the given command line arguments.
Args:
cli_args: An object containing the command line arguments.
"""
with open(cli_args.visualization_config, "r") as f:
viz_config = yaml.safe_load(f)
if cli_args.json_state:
target_directory = (
cli_args.json_state.rsplit(".", maxsplit=1)[0]
if cli_args.output == "<json_state_name>"
else cli_args.output
)
from_json_states(
cli_args.json_state,
viz_config,
target_directory,
cli_args.player_id,
cli_args.display,
cli_args.grid_size,
)
elif cli_args.video:
target_directory = (
cli_args.video + ".mp4"
if cli_args.output == "<json_state_name>"
else cli_args.output
)
video_from_images(cli_args.video, target_directory, cli_args.fps)
else:
target_directory = (
cli_args.action_recording.rsplit(".", maxsplit=1)[0]
if cli_args.output == "<json_state_name>"
else cli_args.output
)
simulate(
cli_args.action_recording,
cli_args.env_configs,
viz_config,
target_directory,
cli_args.player_id,
cli_args.step_duration,
cli_args.fps,
cli_args.display,
cli_args.number_player,
cli_args.break_when_no_action,
cli_args.grid_size,
)
if __name__ == "__main__":
parser = ArgumentParser(
prog="Cooperative Cuisine Video Generation",
description="Generate videos from recorded data.",
epilog="For further information, see https://scs.pages.ub.uni-bielefeld.de/cocosy/overcooked-simulator/overcooked_simulator.html",
)
create_replay_parser(parser)
args = parser.parse_args()
main(args)