-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathCPS.py
More file actions
94 lines (81 loc) · 3.31 KB
/
Copy pathCPS.py
File metadata and controls
94 lines (81 loc) · 3.31 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
# Basic Template generated by x0prc. Modify according to usage.
import numpy as np
import random
from collections import deque
from keras.models import Sequential
from keras.layers import Dense
from keras.optimizers import Adam
# Define parameters and constants
EPISODES = 1000
MAX_STEPS = 100
GAMMA = 0.95
EPSILON = 1.0
EPSILON_MIN = 0.01
EPSILON_DECAY = 0.995
LEARNING_RATE = 0.001
class DRLAgent:
def __init__(self, state_size, action_size):
self.state_size = state_size
self.action_size = action_size
self.memory = deque(maxlen=2000) # Replay memory
self.gamma = GAMMA # Discount rate
# Exploration-exploitation parameters
self.epsilon = EPSILON
self.epsilon_min = EPSILON_MIN
self.epsilon_decay = EPSILON_DECAY
# Neural network model
self.model = self._build_model()
def _build_model(self):
model = Sequential()
model.add(Dense(24, input_dim=self.state_size, activation='relu'))
model.add(Dense(24, activation='relu'))
model.add(Dense(self.action_size, activation='linear'))
model.compile(loss='mse', optimizer=Adam(lr=LEARNING_RATE))
return model
def remember(self, state, action, reward, next_state, done):
self.memory.append((state, action, reward, next_state, done))
def act(self, state):
if np.random.rand() <= self.epsilon:
return random.randrange(self.action_size)
act_values = self.model.predict(state)
return np.argmax(act_values[0])
def replay(self, batch_size):
minibatch = random.sample(self.memory, batch_size)
for state, action, reward, next_state, done in minibatch:
target = reward
if not done:
target = (reward + self.gamma *
np.amax(self.model.predict(next_state)[0]))
target_f = self.model.predict(state)
target_f[0][action] = target
self.model.fit(state, target_f, epochs=1, verbose=0)
if self.epsilon > self.epsilon_min:
self.epsilon *= self.epsilon_decay
# Main function for training
def train_drl_agent():
# Initialize environment and agent
state_size = # Define state size based on environment
action_size = # Define action size based on environment
agent = DRLAgent(state_size, action_size)
# Iterate over episodes
for e in range(EPISODES):
state = # Reset environment and get initial state
for step in range(MAX_STEPS):
# Choose action
action = agent.act(state)
# Perform action and observe reward and next state
next_state, reward, done, _ = # Implement environment step function
# Store experience in replay memory
agent.remember(state, action, reward, next_state, done)
# Update state
state = next_state
# Perform experience replay
if len(agent.memory) > batch_size:
agent.replay(batch_size)
if done:
break
# Decay exploration rate after each episode
if agent.epsilon > agent.epsilon_min:
agent.epsilon *= agent.epsilon_decay
if __name__ == "__main__":
train_drl_agent()