SocialAISchool / scripts /create_LLM_examples_old.py
grg's picture
Cleaned old git history
be5548b
#!/usr/bin/env python3
import time
import argparse
import numpy as np
import gym
import gym_minigrid
from gym_minigrid.wrappers import *
from gym_minigrid.window import Window
from utils import *
from models import MultiModalBaby11ACModel
from collections import Counter
import torch_ac
import json
from termcolor import colored, COLORS
from functools import partial
from tkinter import *
from torch.distributions import Categorical
inter_acl = False
draw_tree = True
def redraw(img):
if not args.agent_view:
img = env.render('rgb_array', tile_size=args.tile_size, mask_unobserved=args.mask_unobserved)
window.show_img(img)
def reset():
# if args.seed != -1:
# env.seed(args.seed)
obs = env.reset()
if hasattr(env, 'mission'):
print('Mission: %s' % env.mission)
window.set_caption(env.mission)
redraw(obs)
tot_bonus = [0]
prev = {
"prev_obs": None,
"prev_info": {},
}
shortened_obj_names = {
'lockablebox' : 'loc_box',
'applegenerator' : 'app_gen',
'generatorplatform': 'gen_pl',
'marbletee' : 'tee',
'remotedoor' : 'rem_door',
}
IDX_TO_OBJECT = {v: shortened_obj_names.get(k, k) for k, v in OBJECT_TO_IDX.items()}
# no duplicates
assert len(IDX_TO_OBJECT) == len(OBJECT_TO_IDX)
IDX_TO_COLOR = {v: k for k, v in COLOR_TO_IDX.items()}
assert len(IDX_TO_COLOR) == len(COLOR_TO_IDX)
# def to_string(enc):
# s = "{:<8} {} {} {} {} {:3} {:3} {}\t".format(
# IDX_TO_OBJECT.get(enc[0], enc[0]), # obj
# *enc[1:3], # x, y
# IDX_TO_COLOR.get(enc[3], enc[3])[:1].upper(), # color
# *enc[4:] #
# )
#
# if IDX_TO_OBJECT.get(enc[0], enc[0]) == "unseen":
# pass
# # s = colored(s, "on_grey")
#
# elif IDX_TO_OBJECT.get(enc[0], enc[0]) != "empty":
# col = IDX_TO_COLOR.get(enc[3], enc[3])
# if col in COLORS:
# s = colored(s, col)
#
# return s
def step(action):
if type(action) == np.ndarray:
obs, reward, done, info = env.step(action)
else:
action = [int(action), np.nan, np.nan]
obs, reward, done, info = env.step(action)
redraw(obs)
if done:
print('done!')
print('Reward=%.2f' % (reward))
print('Exploration_bonus=%.2f' % (tot_bonus[0]))
tot_bonus[0] = 0
with open(output_file, "a") as f:
if reward > 0:
f.write("Success!\n")
f.write("New episode.\n")
reset()
else:
print('\nStep=%s' % (env.step_count))
# print to screen
print("Obs : ", end="")
print("".join(info["descriptions"]), end="")
if obs["utterance_history"] != "Conversation: \n":
print(obs['utterance_history'])
print("Act : ", end="")
# write to file
with open(output_file, "a") as f:
f.write("Obs : ")
f.write("".join(info["descriptions"]))
if obs["utterance_history"] != "Conversation: \n":
f.write(obs['utterance_history'])
# f.write("Your possible actions are:\n")
# f.write("(a) move forward\n")
# f.write("(b) turn left\n")
# f.write("(c) turn right\n")
# f.write("(d) toggle\n")
# f.write("(e) no_op\n")
f.write("Act : ")
print('Full reward (undiminshed)=%.2f' % (reward))
def key_handler(event):
# if hasattr(event.canvas, "_event_loop") and event.canvas._event_loop.isRunning():
# return
print('pressed', event.key)
action_dict = {
"up": "a) move forward",
"left": "b) turn left",
"right": "c) turn right",
" ": "d) toggle",
"shift": "e) no_op",
}
action_dict = {
"up": "move forward",
"left": "turn left",
"right": "turn right",
" ": "toggle",
"shift": "no_op",
}
if event.key in action_dict:
your_action = action_dict[event.key]
with open(output_file, "a") as f:
f.write("{}\n".format(your_action))
if event.key == 'escape':
window.close()
return
if event.key == 'r':
reset()
return
if event.key == 'tab':
step(np.array([np.nan, np.nan, np.nan]))
return
if event.key == 'shift':
step(np.array([np.nan, np.nan, np.nan]))
return
if event.key == 'left':
step(env.actions.left)
return
if event.key == 'right':
step(env.actions.right)
return
if event.key == 'up':
step(env.actions.forward)
return
if event.key == 't':
step(env.actions.speak)
return
if event.key == '1':
step(np.array([np.nan, 0, 0]))
return
if event.key == '2':
step(np.array([np.nan, 0, 1]))
return
if event.key == '3':
step(np.array([np.nan, 1, 0]))
return
if event.key == '4':
step(np.array([np.nan, 1, 1]))
return
if event.key == '5':
step(np.array([np.nan, 2, 2]))
return
if event.key == '6':
step(np.array([np.nan, 1, 2]))
return
if event.key == '7':
step(np.array([np.nan, 2, 1]))
return
if event.key == '8':
step(np.array([np.nan, 1, 3]))
return
if event.key == 'p':
step(np.array([np.nan, 3, 3]))
return
# Spacebar
if event.key == ' ':
step(env.actions.toggle)
return
if event.key == '9':
step(env.actions.pickup)
return
if event.key == '0':
step(env.actions.drop)
return
if event.key == 'enter':
step(env.actions.done)
return
parser = argparse.ArgumentParser()
parser.add_argument(
"--env",
help="gym environment to load",
# default="SocialAI-AsocialBoxInformationSeekingParamEnv-v1",
default="SocialAI-ColorBoxesLLMCSParamEnv-v1",
)
parser.add_argument(
"--seed",
type=int,
help="random seed to generate the environment with",
default=-1
)
parser.add_argument(
"--tile_size",
type=int,
help="size at which to render tiles",
default=32
)
parser.add_argument(
'--agent_view',
default=False,
help="draw the agent sees (partially observable view)",
action='store_true'
)
parser.add_argument(
'--print_grid',
default=False,
help="print the grid with symbols",
action='store_true'
)
parser.add_argument(
'--calc-bonus',
default=False,
help="calculate explo bonus",
action='store_true'
)
parser.add_argument(
'--mask-unobserved',
default=False,
help="mask cells that are not observed by the agent",
action='store_true'
)
parser.add_argument(
'--output-file',
default="./llm_data/in_context_color_test.txt",
help="file where to save episodes",
)
# Put all env related arguments after --env_args, e.g. --env_args nb_foo 1 is_bar True
parser.add_argument("--env-args", nargs='*', default=None)
args = parser.parse_args()
output_file=args.output_file
env = gym.make(args.env, **env_args_str_to_dict(args.env_args))
if draw_tree:
# draw tree
env.parameter_tree.draw_tree(
filename="viz/SocialAIParam/{}_raw_tree".format(args.env),
ignore_labels=["Num_of_colors"],
)
if args.seed >= 0:
env.seed(args.seed)
with open(output_file, "a") as f:
f.write("New episode.\n")
window = Window('gym_minigrid - ' + args.env, figsize=(4, 4))
window.reg_key_handler(key_handler)
env.window = window
# Blocking event loop
window.show(block=True)