Maze

This example shows how OptFlow optimizes the route of a player to escape from a maze

# Author: Anonymized for paper review
# Case Study: Maze

import optflow as flow
import numpy as np

N = 5
x_init = flow.Input(dtype=flow.categorical, size=N)
y_init = flow.Input(dtype=flow.categorical, size=N)
move = flow.Variable(cat=flow.categorical, size=4, shape=(flow.infinite, ))
obj = flow.Constant(value=0.)
maze = flow.Constant(value=np.array([
    [1, 1, 1, 1, 1],
    [1, 0, 0, 0, 0],
    [1, 1, 1, 1, 1],
    [1, 0, 0, 0, 1],
    [1, 0, 1, 1, 1],
]))

x, y = x_init, y_init
constraints = []
with flow.ForLoop(flow.infinite) as t:
    with flow.IfCond(move[t] == 0):     # move down
        x_next = flow.minimum(x + 1, N - 1)
        constraints.append(maze[x_next, y])
        x = x_next
    with flow.IfCond(move[t] == 1):     # move right
        y_next = flow.minimum(y + 1, N - 1)
        constraints.append(maze[x, y_next])
        y = y_next
    with flow.IfCond(move[t] == 2):     # move up
        x_next = flow.maximum(x - 1, 0)
        constraints.append(maze[x_next, y])
        x = x_next
    with flow.IfCond(move[t] == 3):     # move left
        y_next = flow.maximum(y - 1, 0)
        constraints.append(maze[x, y_next])
        y = y_next
    obj -= 1
    with flow.IfCond((x == 0) & (y == 0)):
        flow.break_loop()

flow.analyze_expr(obj, view=True)
prob = flow.Problem(constraints=constraints, objective=obj, sense=flow.maximize)

prob.train(trainer=flow.dp)
opt_obj = prob.solve(solver=flow.dp, init_state_inputs={x_init: 4, y_init: 2})
print(opt_obj)
direction = ['↓', '→', '↑', '←']
for move_i in move.optimized_value:
    print(direction[move_i], end=" ")
print()
opt_obj = prob.solve(solver=flow.dp, init_state_inputs={x_init: 4, y_init: 0})
print(opt_obj)
for move_i in move.optimized_value:
    print(direction[move_i], end=" ")

class DataGenerator(flow.DataGenerator):
    def __init__(self):
        self.available_coords = list(zip(*np.where(maze.value)))

    def __call__(self, *args, **kwargs):
        x, y = self.available_coords[np.random.randint(0, len(self.available_coords))]
        return {x_init: x, y_init: y}

prob.train(trainer=flow.rl, inputs_generator=DataGenerator())
opt_obj_rl = prob.solve(solver=flow.rl, inputs={x_init: 4, y_init: 2})
print(opt_obj_rl)
for move_i in move.optimized_value:
    print(direction[move_i], end=" ")

[Run Online Demo] (password: )