iframe-proxy | Sunbelt Computer Software

433 lines (271 loc) · 10.6 KB
import torch
import numpy as np
import argparse
#from parameters import *
#from PPO import Ppo
from collections import deque
import torch.nn as nn
import torch.optim as optim
import time
import torch.nn.functional as F
from torch.distributions import Normal, kl_divergence
os.add_dll_directory("C:\\Users\\11339\\.mujoco\\mjpro150\\bin")
lr_actor = 0.0003
lr_critic = 0.0003
Iter = 15000
MAX_STEP = 10000
gamma =0.98
lambd = 0.98
batch_size = 64
epsilon = 0.2
l2_rate = 0.001
class Actor(nn.Module):
    def __init__(self,N_S,N_A):
        super(Actor,self).__init__()
        self.fc1 = nn.Linear(N_S,64)
        self.fc2 = nn.Linear(64,64)
        self.sigma = nn.Linear(64,N_A)
        self.mu = nn.Linear(64,N_A)
        #再缩小10倍，且立即替换原值
        self.mu.weight.data.mul_(0.1)
        self.mu.bias.data.mul_(0.0)
        # self.set_init([self.fc1,self.fc2, self.mu, self.sigma])
        #self.distribution = torch.distributions.Normal
    #初始化网络参数
    def set_init(self,layers):
        for layer in layers:
            nn.init.normal_(layer.weight,mean=0.,std=0.1)
            nn.init.constant_(layer.bias,0.)
    def forward(self,s):
        x = torch.tanh(self.fc1(s))
        x = torch.tanh(self.fc2(x))
        mu = self.mu(x)
        log_sigma = self.sigma(x)
        #log_sigma = torch.zeros_like(mu)
        sigma = torch.exp(log_sigma)
        return mu,sigma
    def choose_action(self,s):
        mu,sigma = self.forward(s)
        #Pi = self.distribution(mu,sigma)
        Pi = Normal(mu, sigma)
        return Pi.sample().numpy()
class Critic(nn.Module):
    def __init__(self,N_S):
        super(Critic,self).__init__()
        self.fc1 = nn.Linear(N_S,64)
        self.fc2 = nn.Linear(64,64)
        self.fc3 = nn.Linear(64,1)
        self.fc3.weight.data.mul_(0.1)
        self.fc3.bias.data.mul_(0.0)
        # self.set_init([self.fc1, self.fc2, self.fc2])
    def set_init(self,layers):
        for layer in layers:
            nn.init.normal_(layer.weight,mean=0.,std=0.1)
            nn.init.constant_(layer.bias,0.)
    def forward(self,s):
        x = torch.tanh(self.fc1(s))
        x = torch.tanh(self.fc2(x))
        values = self.fc3(x)
        return values
    def __init__(self,N_S,N_A):
        self.actor_net =Actor(N_S,N_A)
        self.critic_net = Critic(N_S)
        self.actor_optim = optim.Adam(self.actor_net.parameters(),lr=lr_actor)
        self.critic_optim = optim.Adam(self.critic_net.parameters(),lr=lr_critic,weight_decay=l2_rate)
        self.critic_loss_func = torch.nn.MSELoss()
    def train(self,memory):
        memory = np.array(memory)
        states = torch.tensor(np.vstack(memory[:,0]),dtype=torch.float32)
        actions = torch.tensor(list(memory[:,1]),dtype=torch.float32)
        rewards = torch.tensor(list(memory[:,2]),dtype=torch.float32)
        values = self.critic_net(states)
        returns,advants = self.get_gae(rewards,values)
        old_mu,old_std = self.actor_net(states)
        #pi = self.actor_net.distribution(old_mu,old_std)
        old_pi = Normal(old_mu,old_std)
        old_log_prob = old_pi.log_prob(actions).sum(1,keepdim=True)
        n = len(states)
        arr = np.arange(n)   #1~n
        for epoch in range(1):
            np.random.shuffle(arr)
            #向下取整，然后从0~50例如
            for i in range(n//batch_size):
                b_index = arr[batch_size*i:batch_size*(i+1)]
                b_states = states[b_index]
                b_advants = advants[b_index].unsqueeze(1)
                b_actions = actions[b_index]
                b_returns = returns[b_index].unsqueeze(1)
                mu,std = self.actor_net(b_states)
                #pi = self.actor_net.distribution(mu,std)
                new_pi = Normal(mu,std)
                new_prob = new_pi.log_prob(b_actions).sum(1,keepdim=True)
                old_prob = old_log_prob[b_index].detach()
                #KL散度正则项
               # KL_penalty = self.kl_divergence(old_mu[b_index],old_std[b_index],mu,std)
                ratio = torch.exp(new_prob-old_prob)
                surrogate_loss = ratio*b_advants
                values = self.critic_net(b_states)
                #critic_loss = self.critic_loss_func(values,b_returns)
                advantage = values - b_returns
                critic_loss = torch.mean(torch.square(advantage))
                self.critic_optim.zero_grad()
                critic_loss.backward()
                self.critic_optim.step()
                ratio = torch.clamp(ratio,1.0-epsilon,1.0+epsilon)
                clipped_loss =ratio*b_advants
                actor_loss = -torch.min(surrogate_loss,clipped_loss).mean()
                #actor_loss = -(surrogate_loss-beta*KL_penalty).mean()
                self.actor_optim.zero_grad()
                actor_loss.backward()
                self.actor_optim.step()
    #计算KL散度
    def kl_divergence(self,old_mu,old_sigma,mu,sigma):
        old_mu = old_mu.detach()
        old_sigma = old_sigma.detach()
        pi_old = Normal(old_mu,old_sigma)
        pi_new = Normal(mu,sigma)
        kl = kl_divergence(pi_old, pi_new)
        kl_mean = torch.mean(kl)
        '''
        kl = torch.log(old_sigma) - torch.log(sigma) + (old_sigma.pow(2) + (old_mu - mu).pow(2)) / \
             (2.0 * sigma.pow(2)) - 0.5
        return kl.sum(1, keepdim=True)
        '''
        return kl_mean
    def get_gae(self,rewards, values):
        rewards = torch.Tensor(rewards)
        returns = torch.zeros_like(rewards)
        advants = torch.zeros_like(rewards)
        running_returns = 0
        previous_value = 0
        running_advants = 0
        for t in reversed(range(0, len(rewards))):
            #计算A_t并进行加权求和
            running_returns = rewards[t] + gamma * running_returns
            running_tderror = rewards[t] + gamma * previous_value - \
                              values.data[t]
            running_advants = running_tderror + gamma * lambd * \
                              running_advants
            returns[t] = running_returns
            previous_value = values.data[t]
            advants[t] = running_advants
        #advants的归一化
        advants = (advants - advants.mean()) / advants.std()
        return returns, advants
parser = argparse.ArgumentParser()
parser.add_argument('--env_name', type=str, default="Ant-v3",
                    help='name of Mujoco environement')
args = parser.parse_args()
env = gym.make(args.env_name)
N_S = env.observation_space.shape[0]
N_A = env.action_space.shape[0]
env.seed(500)
torch.manual_seed(500)
np.random.seed(500)
class Nomalize:
    def __init__(self, N_S):
        self.mean = np.zeros((N_S,))
        self.std = np.zeros((N_S, ))
        self.stdd = np.zeros((N_S, ))
        self.n = 0
    #可以像函数一样调用类
    def __call__(self, x):
        x = np.asarray(x)
        self.n += 1
        #print("---self.n_____",self.n),此值每调用一次会累加1
        if self.n == 1:
            self.mean = x
        else:
            #更新样本均值和方差
            old_mean = self.mean.copy()
            #print("---old_mean",old_mean)
            self.mean = old_mean + (x - old_mean) / self.n
            #print("-----self.mean:", self.mean)  ，此值会不断累加一个微小的偏移
            self.stdd = self.stdd + (x - old_mean) * (x - self.mean)
        #状态归一化
        if self.n > 1:
            self.std = np.sqrt(self.stdd / (self.n - 1))
        else:
            self.std = self.mean
        x = x - self.mean
        x = x / (self.std + 1e-8)
        x = np.clip(x, -5, +5)
        return x
ppo = Ppo(N_S,N_A)
nomalize = Nomalize(N_S)
#weights_c = torch.load('./model/ppo_critic_{}'.format(ep1))
#ppo.critic_net.load_state_dict(weights_c)
DEVICE = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
s = nomalize(env.reset())
outdir = "./record/dqn-%s" % args.env_name
from gym import wrappers
#env = wrappers.Monitor(env, directory=outdir, force=True)
for ep in range(500):  # 5000 UP
    #s = env.reset()
    #s = torch.from_numpy(s).unsqueeze(dim=0).to(dtype=torch.float32, device=DEVICE)
    weights_a = torch.load('./model/ppo_actor_up_{}'.format(ep1))
    ppo.actor_net.load_state_dict(weights_a)
    #t0 = time.time()
    env.render()
    time.sleep(0.05)
    # PPO：choose action
    a = ppo.actor_net.choose_action(torch.from_numpy(np.array(s).astype(np.float32)).unsqueeze(0))[0]
    s_, r, done, info = env.step(a)
    s_ = nomalize(s_)
    #print("---", ep)
    #s = torch.from_numpy(s_).unsqueeze(dim=0).to(dtype=torch.float32, device=DEVICE)
for ep in range(500):  # 5000 RIGHT
    # s = env.reset()
    # s = torch.from_numpy(s).unsqueeze(dim=0).to(dtype=torch.float32, device=DEVICE)
    weights_a = torch.load('./model/ppo_actor_right_{}'.format(ep1))
    ppo.actor_net.load_state_dict(weights_a)
    # t0 = time.time()
    env.render()
    time.sleep(0.05)
    # PPO：choose action
    a = ppo.actor_net.choose_action(torch.from_numpy(np.array(s).astype(np.float32)).unsqueeze(0))[0]
    s_, r, done, info = env.step(a)
    s_ = nomalize(s_)
    # print("---", ep)
    # s = torch.from_numpy(s_).unsqueeze(dim=0).to(dtype=torch.float32, device=DEVICE)
for ep in range(500):  # 5000  DOWN
    # s = env.reset()
    # s = torch.from_numpy(s).unsqueeze(dim=0).to(dtype=torch.float32, device=DEVICE)
    weights_a = torch.load('./model/ppo_actor_down_{}'.format(ep1))
    ppo.actor_net.load_state_dict(weights_a)
    # t0 = time.time()
    env.render()
    time.sleep(0.05)
    # PPO：choose action
    a = ppo.actor_net.choose_action(torch.from_numpy(np.array(s).astype(np.float32)).unsqueeze(0))[0]
    s_, r, done, info = env.step(a)
    s_ = nomalize(s_)
    # print("---", ep)
    # s = torch.from_numpy(s_).unsqueeze(dim=0).to(dtype=torch.float32, device=DEVICE)
for ep in range(500):  # 5000 LEFT
    #s = env.reset()
    #s = torch.from_numpy(s).unsqueeze(dim=0).to(dtype=torch.float32, device=DEVICE)
    weights_a = torch.load('./model/ppo_actor_left_{}'.format(ep1))
    ppo.actor_net.load_state_dict(weights_a)
    #t0 = time.time()
    env.render()
    time.sleep(0.05)
    # PPO：choose action
    a = ppo.actor_net.choose_action(torch.from_numpy(np.array(s).astype(np.float32)).unsqueeze(0))[0]
    s_, r, done, info = env.step(a)
    s_ = nomalize(s_)
    #print("---", ep)
    #s = torch.from_numpy(s_).unsqueeze(dim=0).to(dtype=torch.float32, device=DEVICE)
Sunbelt Computer Software

PL/B Language Development and Support

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Sunbelt Computer Software

PL/B Language Development and Support

FilesExpand file tree

testAll.py

Latest commit

History

testAll.py

File metadata and controls
