I use the dueling double DQN for a problem in which multiple actions should be selected, i.e., the output of the choose_action function is an action vector selected from an interval of zero and n_actions.
How can I change the codes to get non-repetitive actions (i.e., an action vector with different actions).
The code is as follows:
import numpy as np
import copy, json, argparse
from scipy.io import *
import torch
import tensorflow as tf
import tensorflow.keras as keras
from tensorflow.keras.optimizers import Adam
from random import random, uniform, choices, randint, sample
class Environ():
def __init__(self, Num, n_actions, Pmax, Noise, BW, Rmin, negative_cost):
self.Num = Num
self.state_dim = self.Num
self.n_actions = n_actions
self.Pmax=Pmax
self.Rmin = Rmin
self.Noise=Noise
self.BW=BW
self.negative_cost=negative_cost
self.bs = complex((500 / 2), (500 / 2))
self.QoS = np.zeros(self.Num)
self.S = (np.zeros(self.Num)).reshape(-1)
def Location(self):
rx = np.zeros(self.Num)
ry = np.zeros(self.Num)
Loc = np.zeros(self.Num, dtype=complex)
for i in range(self.Num):
rx[i] = uniform(0, 500)
ry[i] = uniform(0, 500)
Loc[i] = complex(rx[i], ry[i])
return Loc
def PathGain(self, Loc):
d = np.zeros(self.Num)
x = np.zeros(self.n_actions)
h = np.zeros((self.Num, self.n_actions))
for i in range(self.Num):
d[i] = abs(Loc[i] - self.bs)
d[i] = d[i] ** (-3)
for k in range(self.n_actions):
u = np.random.rand(1, 1)
sigma = 1
x[k] = sigma * np.sqrt(-2 * np.log(u))
h[i, k] = d[i] * x[k]
return h
def reset(self): # Reset the states
s = np.zeros(self.Num)
return s.reshape(-1)
def RecievePower(self, h):
TotalPower = np.zeros(self.Num, dtype=float)
P_private = np.zeros((self.Num, self.n_actions), dtype=float)
P_common = np.zeros((self.Num, self.n_actions), dtype=float)
UsersRecievePower_private = np.zeros((self.Num, self.n_actions), dtype=float)
UsersRecievePower_common = np.zeros((self.Num, self.n_actions), dtype=float)
actionPC=np.random.uniform(size=2*self.n_actions)
actionPC=actionPC*self.Pmax
for i in range(self.Num):
for k in range(self.n_actions):
for s in range(i, (i + 1)):
P_private[i, k] = actionPC[s]
P_common[i, k] = actionPC[s + self.n_actions]
for j in range(self.Num):
if j != i :
for l in range(j, (j + 1)):
P_private[i, k] = actionPC[l]
P_common[i, k] = actionPC[l + self.n_actions]
TotalPower[i] = sum(P_private[i, :]) + sum(P_common[i, :])
for i in range(self.Num):
for k in range(self.n_actions):
UsersRecievePower_private[i, k] = h[i, k] * P_private[i, k]
UsersRecievePower_common[i, k] = h[i, k] * P_common[i, k]
return UsersRecievePower_private, UsersRecievePower_common, TotalPower
def TotalRate(self, actionRB,h):
interference_common = np.zeros((self.Num, self.n_actions), dtype=float) + self.Noise
interference_private = np.zeros((self.Num, self.n_actions), dtype=float) + self.Noise
SINR_common = np.zeros((self.Num, self.n_actions), dtype=float)
SINR_private = np.zeros((self.Num, self.n_actions), dtype=float)
Rate_common = np.zeros((self.Num, self.n_actions), dtype=float)
Rate_private = np.zeros((self.Num, self.n_actions), dtype=float)
TotalRate = np.zeros(self.Num, dtype=float)
RecievePower_private, RecievePower_common, TotalPower = self.RecievePower(h)
RB = np.zeros((self.Num, self.n_actions), dtype=float)
for i in range(self.Num):
for k in range(self.n_actions):
for s in range(i, (i + 1)):
if k == int(actionRB[s]):
RB[i, k] = 1
for j in range(self.Num):
if j != i :
for l in range(j, (j + 1)):
if k == int(actionRB[l]):
RB[i, k] = 1
for i in range(self.Num):
for k in range(self.n_actions):
for j in range(self.Num):
if j != i and h[j, k] > h[i, k]:
interference_common[i, k] = interference_common[i, k] + RecievePower_common[j, k]
interference_private[i, k] = interference_private[i, k] + RecievePower_private[j, k]
else:
interference_common[i, k] = interference_common[i, k]
interference_private[i, k] = interference_private[i, k]
SINR_common[i, k] = RecievePower_common[i, k] / interference_common[i, k]
SINR_private[i, k] = RecievePower_private[i, k] / interference_private[i, k]
if RB[i, k] == 1:
Rate_common[i, k] = self.BW * (np.log2(1 + SINR_common[i, k]))
Rate_private[i, k] = self.BW * (np.log2(1 + SINR_private[i, k]))
else:
Rate_common[i, k] = 0
Rate_private[i, k] = 0
TotalRate[i] = sum(Rate_private[i, :]) + sum(Rate_common[i, :])
return TotalRate, TotalPower
def computeQoS(self, actionRB, h):
TotalRate, TotalPower = self.TotalRate(actionRB,h)
for i in range(self.Num):
if TotalRate[i] >= self.Rmin :
self.QoS[i] = (1.0)
else:
self.QoS[i] = (0.0)
return self.QoS
def ComputeState(self, actionRB, h):
self.QoS = self.computeQoS(actionRB, h)
S = np.zeros(self.Num)
for i in range(self.Num):
S[i] = self.QoS[i]
self.S = S
return self.S.reshape(-1)
def Reward(self, actionRB, h):
Rate, Power = self.TotalRate(actionRB, h)
Satisfied_Users = sum(self.QoS)
TotalRate = 0.0
TotalPower = 0.05
for i in range(self.Num):
TotalRate = TotalRate + Rate[i]
TotalPower = TotalPower + Power[i]
FeMBB_TotalPower_Final = {}
if Satisfied_Users == self.Num:
reward = TotalRate / TotalPower
done = True
else:
reward = self.negative_cost
done = False
return reward, done
def step(self, actionRB, h):
next_s = self.ComputeState(actionRB, h)
r, d = self.Reward(actionRB, h)
done = False
info = None
if d == True:
done = True
return next_s, r, done, info
class ReplayBuffer(object):
def __init__(self, max_mem, input_shape, n_actions):
self.memorySize=max_mem
self.couter=0
self.state_transition = np.zeros((self.memorySize, *input_shape), dtype=np.float32)
self.newstate_transition = np.zeros((self.memorySize, *input_shape), dtype=np.float32)
self.action_transition=np.zeros((self.memorySize, n_actions), dtype=np.int32)
self.reward_transition = np.zeros(self.memorySize, dtype=np.float32)
self.terminal_transition=np.zeros(self.memorySize, dtype=np.bool)
def store_transition(self, state, newstate, action, reward, done):
index=self.couter % self.memorySize
self.state_transition[index]=state
self.newstate_transition[index]=newstate
self.action_transition[index] = action
self.reward_transition[index] = reward
self.terminal_transition[index] = done
self.couter+=1
def sample_transition(self, batch_size):
max_mem=min(self.couter, self.memorySize)
batch=np.random.choice(max_mem, batch_size, replace=False)
state=self.state_transition[batch]
newstate=self.newstate_transition[batch]
action=self.action_transition[batch]
reward=self.reward_transition[batch]
done=self.terminal_transition[batch]
return state, newstate, action, reward, done
class D3QN(keras.Model):
def __init__(self, n_actions, layer1_dim, layer2_dim):
super(D3QN, self).__init__()
self.L1 = keras.layers.Dense(layer1_dim, activation='relu')
self.L2 = keras.layers.Dense(layer2_dim, activation='relu')
self.V = keras.layers.Dense(1, activation=None)
self.A = keras.layers.Dense(n_actions, activation=None)
def call(self,state):
x = self.L1(state)
x = self.L2(x)
V = self.V(x)
A = self.A(x)
Q = V + (A - tf.math.reduce_mean(A, axis=1, keepdim=True))
return Q
def advantage(self, state):
x = self.L1(state)
x = self.L2(x)
A = self.A(x)
return A
class D3QNAgent():
def __init__(self, lr, gamma, n_actions,epsilon, batch_size, input_shape,
eps_decay=1e-3, eps_end=0.01, mem_size=1000000, layer1_size=256,
layer2_size=256, replace=100):
self.action_space=[i for i in range(n_actions)]
self.lr=lr
self.gamma=gamma
self.n_actions=n_actions
self.epsilon=epsilon
self.batch_size=batch_size
self.input_shape=input_shape
self.eps_decay=eps_decay
self.eps_end=eps_end
self.mem_size=mem_size
self.layer1_size=layer1_size
self.layer2_size=layer2_size
self.replace=replace
self.learn_step_counter=0
self.memory = ReplayBuffer(self.mem_size, self.input_shape, self.n_actions)
self.q_eval = D3QN(self.n_actions, self.layer1_size, self.layer2_size)
self.q_next = D3QN(self.n_actions, self.layer1_size, self.layer2_size) # Q target
self.q_eval.compile(optimizer=Adam(learning_rate=lr), loss='mean_squared_error')
self.q_next.compile(optimizer=Adam(learning_rate=lr), loss='mean_squared_error')
def store_transition(self, state, newstate, action, reward, done):
self.memory.store_transition(state,newstate,action,reward,done)
def choose_action(self, observation):
actions_total = []
state = np.array([observation])
if np.random.random() < self.epsilon:
actions_total = sample(range(self.n_actions), self.n_actions)
else:
for i in range(self.n_actions):
actions = self.q_eval.advantage(state)
action = tf.math.argmax(actions, axis=1).numpy()[0]
actions_total.append(action)
return actions_total
def learn(self):
if self.learn_step_counter < self.batch_size:
return
if self.learn_step_counter % self.replace ==0:
self.q_next.set_weights(self.q_eval.get_weights())
states, newstates, actions, rewards, dones= self.memory.sample_transition(self.batch_size)
q_pred=self.q_eval(states)
q_next=self.q_next(newstates)
q_target=q_pred.numpy() # y
max_action=tf.math.argmax(self.q_eval(newstates),axis=1)
for idx, done in enumerate(dones):
q_target[idx, actions[idx]] = rewards[idx] + self.gamma * (
1 - int(dones[idx]) * q_next[idx, max_action[idx]])
self.q_eval.train_on_batch(states, q_target)
if self.epsilon > self.eps_end:
self.epsilon=self.epsilon-self.eps_decay
else:
self.epsilon=self.eps_end
self.learn_step_counter+=1
if __name__=='__main__':
device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
parser = argparse.ArgumentParser()
parser.add_argument('--n_actions', type=int, default=4, help='Number of actions')
parser.add_argument('--Num', type=int, default=2, help='Number of users')
parser.add_argument('--learningrate', type=float, default=0.0005, help='Learning rate')
parser.add_argument('--eps', type=float, default=0.9, help='epsilon')
parser.add_argument('--Pmax', type=float, default=0.01, help='maximum transmit power')
parser.add_argument('--Noise', type=float, default=0.00000000000001, help='Noise')
parser.add_argument('--negative_cost', type=float, default=-1.0, help='negative cost for panalty')
parser.add_argument('--Rmin', type=float, default=1000000, help='Minimum QoS')
parser.add_argument('--BW', type=float, default=180000, help='Bandwidth')
parser.add_argument('--gamma', type=float, default=0.99, help='discount factor')
parser.add_argument('--nepisodes', type=int, default=20, help='Number of episodes')
parser.add_argument('--batch_size', type=int, default=64, help='Batch size')
parser.add_argument('--nsteps', type=int, default=50, help='Number of steps')
parser.add_argument('--layer1_size', type=int, default=256)
parser.add_argument('--layer2_size', type=int, default=256)
args = parser.parse_args()
env = Environ(Num=args.Num, n_actions=args.n_actions, Pmax=args.Pmax, Noise=args.Noise, BW=args.BW, Rmin=args.Rmin, negative_cost=args.negative_cost )
agent= D3QNAgent( lr=args.learningrate, gamma=args.gamma,
n_actions=args.n_actions, epsilon=args.eps,
batch_size=args.batch_size,
input_shape=[args.Num])
Total_EE=[]
eps_history =[]
episodes_positive = 0
for i in range(args.nepisodes):
Loc = env.Location()
h = env.PathGain(Loc)
done = False
observation = env.reset()
Reward=0
nstep=0
while not done and nstep<=args.nsteps:
nstep+=1
action = agent.choose_action(observation)
print('action: ', action)
new_observation , reward , done , infor = env.step(action, h )
agent.store_transition(observation , new_observation , action, reward , int(done ))
agent.learn()
observation = new_observation
Reward=reward
if Reward>0 :
episodes_positive +=1
Total_EE.append(Reward )
print('Max EE: ', max(Total_EE)/10**9)
print('Avg EE: ', np.mean(Total_EE)/10**9)
I appreciate any help in solving this issue.