Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def compute_closest_fluents(self, unreachable_fluent_vec, known_action_seqs):
"""
computes the closest fluents to the unreachable fluent vec
:param unreachable_fluent_vec: the unreachable (target) fluent vec
:param known_action_seqs: reachable fluent vecs and their action sequences
:return:
"""
starting_fluent_vec = []
distances = []
for known_action_seq in list(known_action_seqs.keys()):
known_fluent_vec = common.delinearize_fluent_vec(
known_action_seq, self.n_fluents
)
dist = self.fluent_dist(unreachable_fluent_vec, known_fluent_vec)
starting_fluent_vec.append(known_action_seq)
distances.append(dist)
# sort according to shortest distance
distances = np.array(distances)
starting_fluent_vec = np.array(starting_fluent_vec)
arg_order = distances.argsort()
distances = distances[arg_order]
starting_fluent_vec = starting_fluent_vec[arg_order]
return distances, starting_fluent_vec
# take action
obs, rew, done, info = env.step(exec_action)
# import time
print(obs['OBJ_STATES'])
print(obs['_FSM_STATE'])
# #time.sleep(5)
# append post-observation entry to results list
i += 1
new_state = create_state_entry(obs, i, col_label, index_map)
results.append(new_state)
else:
raise ValueError('whoops that is not a valid action!')
# determine if fluent state matches the unobserved state (action sequence is a success), and save successful paths as known_paths
unobserved_fluent_vec = causality.common.delinearize_fluent_vec(unobserved_fluent, causal_planner.n_fluents)
final_fluent_vec = np.array(new_state[1:5])
# we reached the desired state
if np.array_equal(unobserved_fluent_vec, final_fluent_vec):
# add the action sequence to the known paths
if unobserved_fluent not in list(causal_planner.known_action_seqs.keys()):
causal_planner.known_action_seqs[unobserved_fluent] = [possible_action_seq]
else:
causal_planner.known_action_seqs[unobserved_fluent].append(possible_action_seq)
# reset the environment for next possible sequence
obs = env.reset()
print('env reset')
print(obs['OBJ_STATES'])
print(obs['_FSM_STATE'])
time.sleep(1)