-
Notifications
You must be signed in to change notification settings - Fork 1
/
Copy pathtester.py
executable file
·76 lines (65 loc) · 2.29 KB
/
tester.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
import numpy as np
from envs import make_env
from policy.replay_buffer import goal_based_process
from utils.os_utils import make_dir
class Tester:
def __init__(self, args):
self.args = args
self.env = make_env(args)
self.env_test = make_env(args)
self.info = []
if args.save_acc:
make_dir('log/accs', clear=False)
self.test_rollouts = 100
self.env_List = []
self.env_test_List = []
for _ in range(self.test_rollouts):
self.env_List.append(make_env(args))
self.env_test_List.append(make_env(args))
self.acc_record = {}
self.acc_record[self.args.goal] = []
for key in self.acc_record.keys():
self.info.append('Success/'+key+'@blue')
def test_acc(self, key, env, agent):
acc_sum, obs = 0.0, []
done = [False for _ in range(self.test_rollouts)]
result = [{} for _ in range(self.test_rollouts)]
for i in range(self.test_rollouts):
obs.append(goal_based_process(env[i].reset()))
for timestep in range(self.args.timesteps):
actions = agent.step_batch(obs)
obs, infos = [], []
for i in range(self.test_rollouts):
ob, _, _, info = env[i].step(actions[i])
if info['Success']==1 and done[i]==False:
done[i] = True
result[i] = info.copy()
elif done[i]==False and timestep==self.args.timesteps-1:
result[i] = info.copy()
obs.append(goal_based_process(ob))
infos.append(info)
if self.args.test_last_step:
for i in range(self.test_rollouts):
acc_sum += infos[i]['Success']
else:
for i in range(self.test_rollouts):
acc_sum += result[i]['Success']
steps = self.args.buffer.counter
acc = acc_sum/self.test_rollouts
self.acc_record[key].append((steps,acc))
self.args.logger.add_record('Success/'+key, acc)
def cycle_summary(self):
if self.args.save_acc:
self.test_acc(self.args.goal, self.env_List, self.args.agent)
def epoch_summary(self):
if self.args.save_acc:
for key, acc_info in self.acc_record.items():
log_folder = 'accs'
if self.args.tag!='': log_folder = log_folder+'/'+self.args.tag
self.args.logger.save_npz(acc_info, key, log_folder)
def final_summary(self):
if self.args.save_acc:
for key, acc_info in self.acc_record.items():
log_folder = 'accs'
if self.args.tag!='': log_folder = log_folder+'/'+self.args.tag
self.args.logger.save_npz(acc_info, key, log_folder)