|
|
|
@ -25,14 +25,17 @@ from query_sample_factory_checkpoint import SampleFactoryNNQueryWrapper |
|
|
|
import time |
|
|
|
|
|
|
|
tempest_binary = "/home/spranger/projects/tempest-devel/ranking_release/bin/storm" |
|
|
|
mdp_file = "simplified.prism" |
|
|
|
rom_file = "/home/spranger/research/Skiing/env/lib/python3.8/site-packages/AutoROM/roms/skiing.bin" |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
class Verdict(Enum): |
|
|
|
INCONCLUSIVE = 1 |
|
|
|
GOOD = 2 |
|
|
|
BAD = 3 |
|
|
|
|
|
|
|
verdict_to_color_map = {Verdict.BAD: "200,0,0", Verdict.INCONCLUSIVE: "40,40,200", Verdict.GOOD: "00,200,100"} |
|
|
|
|
|
|
|
def convert(tuples): |
|
|
|
return dict(tuples) |
|
|
|
|
|
|
|
@ -101,7 +104,7 @@ def drawImportantStates(important_states): |
|
|
|
|
|
|
|
ski_position_counter = {1: (Action.LEFT, 40), 2: (Action.LEFT, 35), 3: (Action.LEFT, 30), 4: (Action.LEFT, 10), 5: (Action.NOOP, 1), 6: (Action.RIGHT, 10), 7: (Action.RIGHT, 30), 8: (Action.RIGHT, 40) } |
|
|
|
def run_single_test(ale, nn_wrapper, x,y,ski_position, duration=200): |
|
|
|
print(f"Running Test from x: {x:04}, y: {y:04}, ski_position: {ski_position}", end="") |
|
|
|
#print(f"Running Test from x: {x:04}, y: {y:04}, ski_position: {ski_position}", end="") |
|
|
|
for i, r in enumerate(ramDICT[y]): |
|
|
|
ale.setRAM(i,r) |
|
|
|
ski_position_setting = ski_position_counter[ski_position] |
|
|
|
@ -130,13 +133,13 @@ def run_single_test(ale, nn_wrapper, x,y,ski_position, duration=200): |
|
|
|
speed_list.append(ale.getRAM()[14]) |
|
|
|
if len(speed_list) > 15 and sum(speed_list[-6:-1]) == 0: |
|
|
|
return (Verdict.BAD, first_action) |
|
|
|
time.sleep(0.005) |
|
|
|
#time.sleep(0.005) |
|
|
|
return (Verdict.INCONCLUSIVE, first_action) |
|
|
|
|
|
|
|
def optimalAction(choices): |
|
|
|
return max(choices.items(), key=operator.itemgetter(1))[0] |
|
|
|
|
|
|
|
def computeStateRanking(): |
|
|
|
def computeStateRanking(mdp_file): |
|
|
|
command = f"{tempest_binary} --prism {mdp_file} --buildchoicelab --buildstateval --prop 'Rmax=? [C <= 1000]'" |
|
|
|
exec(command) |
|
|
|
|
|
|
|
@ -160,18 +163,49 @@ def fillStateRanking(file_name, match=""): |
|
|
|
return state_ranking |
|
|
|
|
|
|
|
except EnvironmentError: |
|
|
|
print("TODO file not available. Exiting.") |
|
|
|
print("Ranking file not available. Exiting.") |
|
|
|
sys.exit(1) |
|
|
|
|
|
|
|
computeStateRanking() |
|
|
|
ranking = fillStateRanking("action_ranking") |
|
|
|
sorted_ranking = sorted(ranking.items(), key=lambda x: x[1].ranking) |
|
|
|
|
|
|
|
fixed_left_states = list() |
|
|
|
fixed_right_states = list() |
|
|
|
fixed_noop_states = list() |
|
|
|
|
|
|
|
def populate_fixed_actions(state, action): |
|
|
|
if action == Action.LEFT: |
|
|
|
fixed_left_states.append(state) |
|
|
|
if action == Action.RIGHT: |
|
|
|
fixed_right_states.append(state) |
|
|
|
if action == Action.NOOP: |
|
|
|
fixed_noop_states.append(state) |
|
|
|
|
|
|
|
def update_prism_file(old_prism_file, new_prism_file): |
|
|
|
fixed_left_formula = "formula Fixed_Left = false " |
|
|
|
fixed_right_formula = "formula Fixed_Right = false " |
|
|
|
fixed_noop_formula = "formula Fixed_Noop = false " |
|
|
|
for state in fixed_left_states: |
|
|
|
fixed_left_formula += f" | (x={state.x}&y={state.y}&ski_position={state.ski_position}) " |
|
|
|
for state in fixed_right_states: |
|
|
|
fixed_right_formula += f" | (x={state.x}&y={state.y}&ski_position={state.ski_position}) " |
|
|
|
for state in fixed_noop_states: |
|
|
|
fixed_noop_formula += f" | (x={state.x}&y={state.y}&ski_position={state.ski_position}) " |
|
|
|
fixed_left_formula += ";\n" |
|
|
|
fixed_right_formula += ";\n" |
|
|
|
fixed_noop_formula += ";\n" |
|
|
|
with open(f'{old_prism_file}', 'r') as file : |
|
|
|
filedata = file.read() |
|
|
|
if len(fixed_left_states) > 0: filedata = re.sub(r"^formula Fixed_Left =.*$", fixed_left_formula, filedata, flags=re.MULTILINE) |
|
|
|
if len(fixed_right_states) > 0: filedata = re.sub(r"^formula Fixed_Right =.*$", fixed_right_formula, filedata, flags=re.MULTILINE) |
|
|
|
if len(fixed_noop_states) > 0: filedata = re.sub(r"^formula Fixed_Noop =.*$", fixed_noop_formula, filedata, flags=re.MULTILINE) |
|
|
|
with open(f'{new_prism_file}', 'w') as file: |
|
|
|
file.write(filedata) |
|
|
|
|
|
|
|
ale = ALEInterface() |
|
|
|
|
|
|
|
|
|
|
|
if SDL_SUPPORT: |
|
|
|
ale.setBool("sound", True) |
|
|
|
ale.setBool("display_screen", True) |
|
|
|
#if SDL_SUPPORT: |
|
|
|
# ale.setBool("sound", True) |
|
|
|
# ale.setBool("display_screen", True) |
|
|
|
|
|
|
|
# Load the ROM file |
|
|
|
ale.loadROM(rom_file) |
|
|
|
@ -184,16 +218,34 @@ x = 70 |
|
|
|
|
|
|
|
nn_wrapper = SampleFactoryNNQueryWrapper() |
|
|
|
|
|
|
|
exec("cp testing_1.png /dev/shm/testing.png") |
|
|
|
for important_state in sorted_ranking[-100:-1]: |
|
|
|
optimal_choice = optimalAction(important_state[1].choices) |
|
|
|
#print(important_state[1].choices, f"\t\tOptimal: {optimal_choice}") |
|
|
|
x = important_state[0].x |
|
|
|
y = important_state[0].y |
|
|
|
ski_pos = model_to_actual(important_state[0].ski_position) |
|
|
|
action_taken = run_single_test(ale,nn_wrapper,x,y,ski_pos, duration=50) |
|
|
|
print(f".... {action_taken}") |
|
|
|
markerSize = 1 |
|
|
|
marker = f"-fill 'rgba(255,204,0,{important_state[1].ranking})' -draw 'point {x},{y} '" |
|
|
|
command = f"convert /dev/shm/testing.png {marker} /dev/shm/testing.png" |
|
|
|
exec(command, verbose=False) |
|
|
|
iteration = 0 |
|
|
|
id = int(time.time()) |
|
|
|
init_mdp = "velocity" |
|
|
|
exec(f"mkdir -p images/testing_{id}") |
|
|
|
exec(f"cp 1_full_scaled_down.png images/testing_{id}/testing_0000.png") |
|
|
|
exec(f"cp {init_mdp}.prism {init_mdp}_000.prism") |
|
|
|
|
|
|
|
markerSize = 1 |
|
|
|
markerList = {1: list(), 2:list(), 3:list(), 4:list(), 5:list(), 6:list(), 7:list(), 8:list()} |
|
|
|
|
|
|
|
while True: |
|
|
|
computeStateRanking(f"{init_mdp}_{iteration:03}.prism") |
|
|
|
ranking = fillStateRanking("action_ranking") |
|
|
|
sorted_ranking = sorted(ranking.items(), key=lambda x: x[1].ranking) |
|
|
|
for important_state in sorted_ranking[-100:-1]: |
|
|
|
optimal_choice = optimalAction(important_state[1].choices) |
|
|
|
#print(important_state[1].choices, f"\t\tOptimal: {optimal_choice}") |
|
|
|
x = important_state[0].x |
|
|
|
y = important_state[0].y |
|
|
|
ski_pos = model_to_actual(important_state[0].ski_position) |
|
|
|
result = run_single_test(ale,nn_wrapper,x,y,ski_pos, duration=50) |
|
|
|
#print(f".... {result}") |
|
|
|
marker = f"-fill 'rgba({verdict_to_color_map[result[0]],0.7})' -draw 'rectangle {x-markerSize},{y-markerSize} {x+markerSize},{y+markerSize} '" |
|
|
|
markerList[ski_pos].append(marker) |
|
|
|
populate_fixed_actions(important_state[0], result[1]) |
|
|
|
for pos, marker in markerList.items(): |
|
|
|
command = f"convert images/testing_{id}/testing_0000.png {' '.join(marker)} images/testing_{id}/testing_{iteration+1:03}_{pos:02}.png" |
|
|
|
exec(command, verbose=False) |
|
|
|
exec(f"montage images/testing_{id}/testing_{iteration+1:03}_*png -geometry +0+0 -tile x1 images/testing_{id}/{iteration+1:03}.png", verbose=False) |
|
|
|
iteration += 1 |
|
|
|
update_prism_file(f"{init_mdp}_{iteration-1:03}.prism", f"{init_mdp}_{iteration:03}.prism") |
