def _get_obs(self):# 获取游戏的状态state = self.game.getGameState()# 小鸟与它前面一对水管中下面那根水管的水平距离dist_to_pipe_horz = state["next_pipe_dist_to_player"]# 小鸟与它前面一对水管中下面那根水管的顶端的垂直距离dist_to_pipe_bottom = state["player_y"] - state["next_pipe_top_y"]# 获取小鸟的水平速度velocity = state['player_vel']# 将这些信息封装成一个数据返回return np.array([dist_to_pipe_horz, dist_to_pipe_bottom, velocity])

def learn(self, writer):# 根据经验进行学习# 确保经验池里的经验以及足够多时，才进行学习assert WARMUP_SIZE >= BATCH_SIZEif len(self.memory) < WARMUP_SIZE:return# 从经验池中选取BATCH_SIZE条经验出来batch = random.sample(self.memory, BATCH_SIZE)batch = Transition(*(zip(*batch)))# 把这些经验都转换位Tensors0 = torch.FloatTensor(batch.state).to(device)a0 = torch.LongTensor(batch.action).to(device).unsqueeze(1)r1 = torch.FloatTensor(batch.reward).to(device)s1 = torch.FloatTensor(batch.next_state).to(device)d1 = torch.LongTensor(batch.done).to(device)# DQN算法q_pred = self.model(s0).gather(1, a0).squeeze()with torch.no_grad():if USE_DBQN:acts = self.model(s1).max(1)[1].unsqueeze(1)q_target = self.target_model(s1).gather(1, acts).squeeze(1)else:q_target = self.target_model(s1).max(1)[0]q_target = r1 + GAMMA * (1 - d1) * q_targetloss = self.loss_func(q_pred, q_target)self.optimizer.zero_grad()loss.backward()self.optimizer.step()# 记录每次学习的losswriter.add_scalar('train/value_loss', loss.item(), self.update_count)self.update_count += 1# 每MODEL_SYNC_COUNT同步一下目标模型if self.update_count % MODEL_SYNC_COUNT == 0:self.target_model.load_state_dict(self.model.state_dict())