diff --git a/pytennis/Images/ball.jpg b/pytennis/Images/ball.jpg new file mode 100644 index 0000000..57348aa Binary files /dev/null and b/pytennis/Images/ball.jpg differ diff --git a/pytennis/Images/ball.png b/pytennis/Images/ball.png new file mode 100644 index 0000000..85d3f1f Binary files /dev/null and b/pytennis/Images/ball.png differ diff --git a/pytennis/Images/padA.png b/pytennis/Images/padA.png new file mode 100644 index 0000000..7e223c7 Binary files /dev/null and b/pytennis/Images/padA.png differ diff --git a/pytennis/Images/padB.png b/pytennis/Images/padB.png new file mode 100644 index 0000000..4e9d6ec Binary files /dev/null and b/pytennis/Images/padB.png differ diff --git a/pytennis/network.py b/pytennis/network.py new file mode 100644 index 0000000..2643a76 --- /dev/null +++ b/pytennis/network.py @@ -0,0 +1,158 @@ +from keras import Sequential, layers +from keras.optimizers import Adam +from keras.layers import Dense +from collections import deque +import numpy as np + + +class Network: + def __init__(self, xmin, xmax, ymin, ymax): + """ + xmin: 150, + xmax: 450, + ymin: 100, + ymax: 600 + """ + + self.StaticDiscipline = { + 'xmin': xmin, + 'xmax': xmax, + 'ymin': ymin, + 'ymax': ymax + } + + def network(self, xsource, ysource=100, Ynew=600, divisor=50): # ysource will always be 100 + """ + For Network A + ysource: will always be 100 + xsource: will always be between xmin and xmax (static discipline) + + For Network B + ysource: will always be 600 + xsource: will always be between xmin and xmax (static discipline) + """ + + while True: + ListOfXsourceYSource = [] + Xnew = np.random.choice([i for i in range( + self.StaticDiscipline['xmin'], self.StaticDiscipline['xmax'])], 1) + #Ynew = np.random.choice([i for i in range(self.StaticDiscipline['ymin'], self.StaticDiscipline['ymax'])], 1) + + source = (xsource, ysource) + target = (Xnew[0], Ynew) + + #Slope and intercept + slope = (ysource - Ynew)/(xsource - Xnew[0]) + intercept = ysource - (slope*xsource) + if (slope != np.inf) and (intercept != np.inf): + break + else: + continue + + #print(source, target) + # randomly select 50 new values along the slope between xsource and xnew (monotonically decreasing/increasing) + XNewList = [xsource] + + if xsource < Xnew: + differences = Xnew[0] - xsource + increment = differences / divisor + newXval = xsource + for i in range(divisor): + + newXval += increment + XNewList.append(int(newXval)) + else: + differences = xsource - Xnew[0] + decrement = differences / divisor + newXval = xsource + for i in range(divisor): + + newXval -= decrement + XNewList.append(int(newXval)) + + # determine the values of y, from the new values of x, using y= mx + c + yNewList = [] + for i in XNewList: + findy = (slope * i) + intercept # y = mx + c + yNewList.append(int(findy)) + + ListOfXsourceYSource = [(x, y) for x, y in zip(XNewList, yNewList)] + + return XNewList, yNewList + + def DefaultToPosition(self, x1, x2=300, divisor=50): + DefaultPositionA = 300 + DefaultPositionB = 300 + XNewList = [] + if x1 < x2: + differences = x2 - x1 + increment = differences / divisor + newXval = x1 + for i in range(divisor): + newXval += increment + XNewList.append(int(np.floor(newXval))) + + else: + differences = x1 - x2 + decrement = differences / divisor + newXval = x1 + for i in range(divisor): + newXval -= decrement + XNewList.append(int(np.floor(newXval))) + return XNewList + + + + +class DQN: + def __init__(self): + self.learning_rate = 0.001 + self.momentum = 0.95 + self.eps_min = 0.1 + self.eps_max = 1.0 + self.eps_decay_steps = 2000000 + self.replay_memory_size = 500 + self.replay_memory = deque([], maxlen=self.replay_memory_size) + n_steps = 4000000 # total number of training steps + self.training_start = 10000 # start training after 10,000 game iterations + self.training_interval = 4 # run a training step every 4 game iterations + self.save_steps = 1000 # save the model every 1,000 training steps + self.copy_steps = 10000 # copy online DQN to target DQN every 10,000 training steps + self.discount_rate = 0.99 + # Skip the start of every game (it's just waiting time). + self.skip_start = 90 + self.batch_size = 100 + self.iteration = 0 # game iterations + self.done = True # env needs to be reset + + self.model = self.DQNmodel() + + return + + def DQNmodel(self): + model = Sequential() + model.add(Dense(64, input_shape=(1,), activation='relu')) + model.add(Dense(64, activation='relu')) + model.add(Dense(10, activation='softmax')) + model.compile(loss='categorical_crossentropy', + optimizer=Adam(lr=self.learning_rate)) + return model + + def sample_memories(self, batch_size): + indices = np.random.permutation(len(self.replay_memory))[:batch_size] + # state, action, reward, next_state, continue + cols = [[], [], [], [], []] + for idx in indices: + memory = self.replay_memory[idx] + for col, value in zip(cols, memory): + col.append(value) + cols = [np.array(col) for col in cols] + return (cols[0], cols[1], cols[2].reshape(-1, 1), cols[3], cols[4].reshape(-1, 1)) + + def epsilon_greedy(self, q_values, step): + self.epsilon = max(self.eps_min, self.eps_max - + (self.eps_max-self.eps_min) * step/self.eps_decay_steps) + if np.random.rand() < self.epsilon: + return np.random.randint(10) # random action + else: + return np.argmax(q_values) # optimal action diff --git a/pytennis/pytennis.gif b/pytennis/pytennis.gif new file mode 100644 index 0000000..cd76cbc Binary files /dev/null and b/pytennis/pytennis.gif differ diff --git a/pytennis/pytennis.py b/pytennis/pytennis.py new file mode 100644 index 0000000..5a62265 --- /dev/null +++ b/pytennis/pytennis.py @@ -0,0 +1,428 @@ +import time +import os +import sys +from network import DQN +from network import Network +import numpy as np +from keras.utils import to_categorical +import tensorflow as tf +import pygame +from pygame.locals import * +from keras import Sequential, layers +from keras.optimizers import Adam +from keras.layers import Dense +from collections import deque +pygame.init() + + + +class tennis: + def __init__(self, fps=50): + self.GeneralReward = False + self.net = Network(150, 450, 150, 650) + self.updateRewardA = 0 + self.updateRewardB = 0 + self.updateIter = 0 + self.lossA = 0 + self.lossB = 0 + self.restart = False + self.iteration = 0 + self.AgentA = DQN() + self.AgentB = DQN() + + # Testing + self.net = Network(150, 450, 150, 650) + self.NetworkA = self.net.network( + 300, ysource=80, Ynew=650) # Network A + self.NetworkB = self.net.network( + 200, ysource=650, Ynew=80) # Network B + + pygame.init() + self.BLACK = (0, 0, 0) + + self.myFontA = pygame.font.SysFont("Times New Roman", 25) + self.myFontB = pygame.font.SysFont("Times New Roman", 25) + self.myFontIter = pygame.font.SysFont('Times New Roman', 25) + + self.FPS = fps + self.fpsClock = pygame.time.Clock() + + self.nextplayer = np.random.choice(['A', 'B']) + + def setWindow(self): + + # set up the window + self.DISPLAYSURF = pygame.display.set_mode((600, 750), 0, 32) + pygame.display.set_caption( + 'REINFORCEMENT LEARNING (DQN) - TABLE TENNIS') + # set up the colors + self.BLACK = (0, 0, 0) + self.WHITE = (255, 255, 255) + self.RED = (255, 0, 0) + self.GREEN = (0, 255, 0) + self.BLUE = (0, 0, 255) + + return + + def display(self): + self.setWindow() + self.DISPLAYSURF.fill(self.WHITE) + pygame.draw.rect(self.DISPLAYSURF, self.BLACK, (50, 100, 500, 550)) + pygame.draw.rect(self.DISPLAYSURF, self.RED, (50, 365, 500, 20)) + return + + def reset(self): + return + + def evaluate_state_from_last_coordinate(self, c): + """ + cmax: 550 + cmin: 50 + + c definately will be between 50 and 550. + """ + if c >= 50 and c <= 550: + return int(c/50 - 1) + else: + return 0 + + def evaluate_action(self, diff): + + if (int(diff) <= 50): + return True + else: + return False + + def randomVal(self, action): + "action is a probability of values between 0 and 1" + val = (action*500) + 50 + return val + + def play(self, action, count=0, play = 'A'): + # play = A implies compute player A's next play. + # play = B implies compute player B's next play. + + if play == 'A': + # playerA should play + if count == 0: + self.NetworkA = self.net.network( + self.ballx, ysource=80, Ynew=650) # Network A + self.bally = self.NetworkA[1][count] + self.ballx = self.NetworkA[0][count] + + if self.GeneralReward == True: + self.playerax = self.randomVal(action) + else: + self.playerax = self.ballx + + + else: + self.ballx = self.NetworkA[0][count] + self.bally = self.NetworkA[1][count] + + obsOne = self.evaluate_state_from_last_coordinate( + int(self.ballx)) # last state of the ball + obsTwo = self.evaluate_state_from_last_coordinate( + int(self.playerbx)) # evaluate player bx + diff = np.abs(self.ballx - self.playerbx) + obs = obsTwo + reward = self.evaluate_action(diff) + done = True + info = str(diff) + + else: + # playerB should play + if count == 0: + self.NetworkB = self.net.network( + self.ballx, ysource=650, Ynew=80) # Network B + self.bally = self.NetworkB[1][count] + self.ballx = self.NetworkB[0][count] + + if self.GeneralReward == True: + self.playerbx = self.randomVal(action) + else: + self.playerbx = self.ballx + + + else: + self.ballx = self.NetworkB[0][count] + self.bally = self.NetworkB[1][count] + + obsOne = self.evaluate_state_from_last_coordinate( + int(self.ballx)) # last state of the ball + obsTwo = self.evaluate_state_from_last_coordinate( + int(self.playerax)) # evaluate player bx + diff = np.abs(self.ballx - self.playerax) + obs = obsTwo + reward = self.evaluate_action(diff) + done = True + info = str(diff) + + return obs, reward, done, info + + def computeLoss(self, reward, loss = 'A'): + # loss = A, implies compute loss of player A, otherwise, compute Player B loss. + if loss == 'A': + if reward == 0: + self.lossA += 1 + else: + self.lossA += 0 + else: + if reward == 0: + self.lossB += 1 + else: + self.lossB += 0 + return + + def execute(self, state, iteration, count, player = 'A'): + if player == 'B': + stateB = state + # Online DQN evaluates what to do + + try: + q_valueB = self.AgentB.model.predict([stateB]) + except: + q_valueB = 0 + actionB = self.AgentB.epsilon_greedy(q_valueB, iteration) + + # Online DQN plays + obsB, rewardB, doneB, infoB = self.play( + action=actionB, count=count, play = 'B') + next_stateB = actionB + + # Let's memorize what just happened + self.AgentB.replay_memory.append( + (stateB, actionB, rewardB, next_stateB, 1.0 - doneB)) + stateB = next_stateB + + output = (q_valueB, actionB, obsB, rewardB, doneB, infoB, next_stateB, actionB, stateB) + + else: + stateA = state + # Online DQN evaluates what to do + # arr = np.array([stateA]) + try: + q_valueA = self.AgentB.model.predict([stateB]) + except: + q_valueA = 0 + actionA = self.AgentA.epsilon_greedy(q_valueA, iteration) + + # Online DQN plays + obsA, rewardA, doneA, infoA = self.play( + action=actionA, count=count, play = 'A') + next_stateA = actionA + + # Let's memorize what just happened + self.AgentA.replay_memory.append( + (stateA, actionA, rewardA, next_stateA, 1.0 - doneA)) + stateA = next_stateA + + output = (q_valueA, actionA, obsA, rewardA, doneA, infoA, next_stateA, actionA, stateA) + + return output + + def trainOnlineDQN(self, player = 'A'): + if player == 'A': + X_state_val, X_action_val, rewards, X_next_state_val, continues = ( + self.AgentA.sample_memories(self.AgentA.batch_size)) + arr = [X_next_state_val] + next_q_values = self.AgentA.model.predict(arr) + max_next_q_values = np.max( + next_q_values, axis=1, keepdims=True) + y_val = rewards + continues * self.AgentA.discount_rate * max_next_q_values + + # Train the online DQN + self.AgentA.model.fit(X_state_val, tf.keras.utils.to_categorical( + X_next_state_val, num_classes=10), verbose=0) + else: + X_state_val, X_action_val, rewards, X_next_state_val, continues = ( + self.AgentB.sample_memories(self.AgentB.batch_size)) + arr = [X_next_state_val] + next_q_values = self.AgentB.model.predict(arr) + max_next_q_values = np.max( + next_q_values, axis=1, keepdims=True) + y_val = rewards + continues * self.AgentB.discount_rate * max_next_q_values + + # Train the online DQN + self.AgentB.model.fit(X_state_val, tf.keras.utils.to_categorical( + X_next_state_val, num_classes=10), verbose=0) + + + return True + + def show_board(self): + self.display() + # CHECK BALL MOVEMENT + self.DISPLAYSURF.blit(self.PLAYERA, (self.playerax, 50)) + self.DISPLAYSURF.blit(self.PLAYERB, (self.playerbx, 650)) + self.DISPLAYSURF.blit(self.ball, (self.ballx, self.bally)) + self.DISPLAYSURF.blit(self.randNumLabelA, (20, 15)) + self.DISPLAYSURF.blit(self.randNumLabelB, (450, 15)) + + + pygame.display.update() + self.fpsClock.tick(self.FPS) + + for event in pygame.event.get(): + + if event.type == QUIT: + # self.AgentA.model.save('models/AgentA.h5') + # self.AgentB.model.save('models/AgentB.h5') + pygame.quit() + sys.exit() + return + + + def step(self, action): + # stepOutput: reward, next_state, done + # action represents the next player to player, action can either be {playerA:0, playerB: 1} + # diplay team players + self.PLAYERA = pygame.image.load('Images/padB.png') + self.PLAYERA = pygame.transform.scale(self.PLAYERA, (50, 50)) + self.PLAYERB = pygame.image.load('Images/padA.png') + self.PLAYERB = pygame.transform.scale(self.PLAYERB, (50, 50)) + self.ball = pygame.image.load('Images/ball.png') + self.ball = pygame.transform.scale(self.ball, (15, 15)) + + self.playerax = 150 + self.playerbx = 250 + + self.ballx = 250 + self.bally = 300 + + # player A starts by playing with state 0 + obsA, rewardA, doneA, infoA = 0, False, False, '' + obsB, rewardB, doneB, infoB = 0, False, False, '' + state = 0 + stateA = 0 + stateB = 0 + next_stateA = 0 + next_stateB = 0 + iteration = self.iteration + actionA = 0 + actionB = 0 + restart = False + + + self.display() + self.randNumLabelA = self.myFontA.render( + 'Score A: '+str(self.updateRewardA), 1, self.BLACK) + self.randNumLabelB = self.myFontB.render( + 'Score B: '+str(self.updateRewardB), 1, self.BLACK) + + nextplayer = self.nextplayer + + if self.nextplayer == 'A': + for count in range(50): + if count == 0: + output = self.execute(state, iteration, count, player = nextplayer) + q_valueA, actionA, obsA, rewardA, doneA, infoA, next_stateA, actionA, stateA = output + state = next_stateA + + + elif count == 49: + + output = self.execute(state, iteration, count, player = 'A') + q_valueA, actionA, obsA, rewardA, doneA, infoA, next_stateA, actionA, stateA = output + state = next_stateA + + self.updateRewardA += rewardA + self.computeLoss(rewardA, loss = 'A') + + # restart the game if player A fails to get the ball, and let B start the game + if rewardA == 0: + self.restart = True + time.sleep(0.5) + self.nextplayer = 'B' + self.GeneralReward = False + else: + self.restart = False + self.GeneralReward = True + + # Sample memories and use the target DQN to produce the target Q-Value + self.trainOnlineDQN(player = 'A') + + self.nextplayer = 'B' + self.updateIter += 1 + + + else: + output = self.execute(state, iteration, count, player = 'A') + q_valueA, actionA, obsA, rewardA, doneA, infoA, next_stateA, actionA, stateA = output + state = next_stateA + + stepOutput = rewardA, next_stateA, doneA + self.show_board() + + else: + for count in range(50): + if count == 0: + output = self.execute(state, iteration, count, player = 'B') + q_valueB, actionB, obsB, rewardB, doneB, infoB, next_stateB, actionB, stateB = output + state = next_stateB + + elif count == 49: + + output = self.execute(state, iteration, count, player = 'B') + q_valueB, actionB, obsB, rewardB, doneB, infoB, next_stateB, actionB, stateB = output + state = next_stateB + + self.updateRewardB += rewardB + self.computeLoss(rewardB, loss = 'B') + + # restart the game if player A fails to get the ball, and let B start the game + if rewardB == 0: + self.restart = True + time.sleep(0.5) + self.GeneralReward = False + self.nextplayer = 'A' + else: + self.restart = False + self.GeneralReward = True + + # Sample memories and use the target DQN to produce the target Q-Value + self.trainOnlineDQN(player = 'B') + + self.nextplayer = 'A' + self.updateIter += 1 + # evaluate B + + else: + output = self.execute(state, iteration, count, player = 'B') + q_valueB, actionB, obsB, rewardB, doneB, infoB, next_stateB, actionB, stateB = output + state = next_stateB + + stepOutput = rewardA, next_stateA, doneA + + self.show_board() + + self.iteration += 1 # keep track of the total number of iterations conducted + return stepOutput + + + +def random_policy(episode): + + action_space = 3 + state_space = 5 + max_steps = 1000 + + for e in range(episode): + state = env.reset() + score = 0 + + for i in range(max_steps): + action = np.random.randint(action_space) + reward, next_state, done = env.step(action) + score += reward + state = next_state + if done: + print("episode: {}/{}, score: {}".format(e, episode, score)) + break + + +if __name__ == "__main__": + env = tennis(fps=70) + action_space = 2 + score = 0 + random_policy(10) \ No newline at end of file