东北大学研究生：用强化学习玩Chrome里的恐龙小游戏

pip install keras
pip install tensorflow

from keras import backend as K
K.tensorflow_backend._get_available_gpus()

pip install selenium

pip install opencv-python

class Game:
    def __init__(self):
        self._driver = webdriver.Chrome(executable_path = chrome_driver_path)
        self._driver.set_window_position(x=-10,y=0)
        self._driver.get(game_url)
    def restart(self):
        self._driver.execute_script("Runner.instance_.restart()")
    def press_up(self):
        self._driver.find_element_by_tag_name("body").send_keys(Keys.ARROW_UP)
    def get_score(self):
        score_array = self._driver.execute_script("return Runner.instance_.distanceMeter.digits")
        score = ''.join(score_array).
        return int(score)

class DinoAgent:
    def __init__(self,game): #takes game as input for taking actions
        self._game = game; 
        self.jump(); #to start the game, we need to jump once
    def is_crashed(self):
        return self._game.get_crashed()
    def jump(self):
        self._game.press_up()

class Game_sate:
    def __init__(self,agent,game):
        self._agent = agent
        self._game = game
    def get_state(self,actions):
        score = self._game.get_score() 
        reward = 0.1 #survival reward
        is_over = False #game over
        if actions[1] == 1: #else do nothing
            self._agent.jump()
        image = grab_screen(self._game._driver) 
        if self._agent.is_crashed():
            self._game.restart()
            reward = -1
            is_over = True
        return image, reward, is_over #return the Experience tuple

#javascript code to get the image data from canvas
var canvas = document.getElementsByClassName('runner-canvas')[0];
var img_data = canvas.toDataURL()
return img_data

def grab_screen(_driver = None):
    image_b64 = _driver.execute_script(getbase64Script)
    screen = np.array(Image.open(BytesIO(base64.b64decode(image_b64))))
    image = process_img(screen)#processing image as required
    return image

def process_img(image):
    image = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
    image = image[:300, :500]
    return image

def buildmodel():
    print("Now we build the model")
    model = Sequential()
    model.add(Conv2D(32, (8, 8), padding='same',strides=(4, 4),input_shape=(img_cols,img_rows,img_channels)))  #80*80*4
    model.add(MaxPooling2D(pool_size=(2,2)))
    model.add(Activation('relu'))
    model.add(Conv2D(64, (4, 4),strides=(2, 2),  padding='same'))
    model.add(MaxPooling2D(pool_size=(2,2)))
    model.add(Activation('relu'))
    model.add(Conv2D(64, (3, 3),strides=(1, 1),  padding='same'))
    model.add(MaxPooling2D(pool_size=(2,2)))
    model.add(Activation('relu'))
    model.add(Flatten())
    model.add(Dense(512))
    model.add(Activation('relu'))
    model.add(Dense(ACTIONS))
    adam = Adam(lr=LEARNING_RATE)
    model.compile(loss='mse',optimizer=adam)
    print("We finish building the model")
    return model

def trainNetwork(model,game_state):
    # store the previous observations in replay memory
    D = deque() #experience replay memory
    # get the first state by doing nothing
    do_nothing = np.zeros(ACTIONS)
    do_nothing[0] =1 #0 => do nothing,
                     #1=> jump
    x_t, r_0, terminal = game_state.get_state(do_nothing) # get next step after performing the action
    s_t = np.stack((x_t, x_t, x_t, x_t), axis=2).reshape(1,20,40,4) # stack 4 images to create placeholder input reshaped 1*20*40*4 
    
    OBSERVE = OBSERVATION
    epsilon = INITIAL_EPSILON
    t = 0
    while (True): #endless running
        
        loss = 0
        Q_sa = 0
        action_index = 0
        r_t = 0 #reward at t
        a_t = np.zeros([ACTIONS]) # action at t
       
        q = model.predict(s_t)       #input a stack of 4 images, get the prediction
        max_Q = np.argmax(q)         # chosing index with maximum q value
        action_index = max_Q 
        a_t[action_index] = 1        # o=> do nothing, 1=> jump
       
        #run the selected action and observed next state and reward
        x_t1, r_t, terminal = game_state.get_state(a_t)
        x_t1 = x_t1.reshape(1, x_t1.shape[0], x_t1.shape[1], 1) #1x20x40x1
        s_t1 = np.append(x_t1, s_t[:, :, :, :3], axis=3) # append the new image to input stack and remove the first one
        
        D.append((s_t, action_index, r_t, s_t1, terminal))# store the transition 
        
        #only train if done observing; sample a minibatch to train on
        trainBatch(random.sample(D, BATCH)) if t > OBSERVE else 0
        s_t = s_t1 
        t += 1

def trainBatch(minibatch):
  for i in range(0, len(minibatch)):
                loss = 0
                inputs = np.zeros((BATCH, s_t.shape[1], s_t.shape[2], s_t.shape[3]))   #32, 20, 40, 4
                targets = np.zeros((inputs.shape[0], ACTIONS))                         #32, 2
                state_t = minibatch[i][0]    # 4D stack of images
                action_t = minibatch[i][1]   #This is action index
                reward_t = minibatch[i][2]   #reward at state_t due to action_t
                state_t1 = minibatch[i][3]   #next state
                terminal = minibatch[i][4]   #wheather the agent died or survided due the action
                inputs[i:i + 1] = state_t    
                targets[i] = model.predict(state_t)  # predicted q values
                Q_sa = model.predict(state_t1)      #predict q values for next step
                if terminal:
                    targets[i, action_t] = reward_t # if terminated, only equals reward
                else:
                    targets[i, action_t] = reward_t + GAMMA * np.max(Q_sa)
            loss += model.train_on_batch(inputs, targets)