# -*- coding: utf-8 -*- """Commented Tensorflow_standing_3(half_success).ipynb Automatically generated by Colaboratory. Original file is located at https://colab.research.google.com/drive/1crNqr9AYW2S4wvDl17h9ZJIQkTBinfoj """ # Szükséges könyvtárak meghívása import tensorflow as tf import tensorflow.contrib.slim as slim import numpy as np import matplotlib.pyplot as plt # %matplotlib inline import time import math try: xrange = xrange except: xrange = range #Környezeti változók inicializálása XX = 800 YY = 800 xflexibility = 1 yflexibility = 1 floordampening = 0.5 floorstickyness = 10 dampening = 0.4 G = 0.001 flor = 50 #Fizikai környezet megahtározásához szükséges objektumok és függvények #L2 norma szerinti távolság def dist(j1, j2): x1 = j1.getx() x2 = j2.getx() y1 = j1.gety() y2 = j2.gety() return math.sqrt((x2-x1)*(x2-x1)+(y2-y1)*(y2-y1)) #ízület osztály class Joint: def __init__(self, x, y, vx, vy, m): self.x = x #pozíciók self.y = y self.vx = vx #sebességek self.vy = vy self.ax = 0 #gyorsulások self.ay = 0 self.m = m #tömeg self.energy = 0 #energia self.prev_x = self.x #előző állapot beli poíciók self.prev_y = self.y #változókat lekérő függvények def getx(self): return self.x def gety(self): return self.y #változókat módosító függvények def repos(self,x, y): self.x = x self.y = y def addacc(self,ax, ay): self.ax += ax self.ay += ay #állapotváltozók iterálása környezeti paraméterek szerint def step(self): #gravitáció hatása self.ay += G*self.m #x irányú visszaverődés megvalósítása (most nem szükséges) #if self.x<=5 or self.x>=XX-5: # self.vx = -self.vx*xflexibility # if self.x<=5: # self.x = 5 # if self.x>=XX-5: # self.x = XX-5 #y irányú visszaverőség megvalósítása if self.y<=5 or self.y>=YY-flor-5: self.vy = -self.vy*yflexibility if self.y<=5: self.y = 5 if self.y>=YY-flor-5: self.y = YY-flor-5 #talaj hatásának megvalósítása if self.y>=YY-flor-7: self.vy = floordampening*self.vy self.vx = floordampening*self.vx if abs(self.vy)0.5: self.l += n if self.l <= 0.01: self.l = 0.01 def strength_modify(self, strength): self.strength = strength def mult_strength(self, n, b): if b>0.5: self.strength *= n #izom viselkedésének leírása (rugó mozgás megvalósítása szabad végű rugóra) def movement(self): d = dist(self.j1, self.j2) ex1 = (self.j2.getx()-self.j1.getx())/d ey1 = (self.j2.gety()-self.j1.gety())/d px1 = (ex1*(d-self.l)/10000*self.strength) py1 = (ey1*(d-self.l)/10000*self.strength) self.j1.addacc(px1,py1) ex2 = (self.j1.getx()-self.j2.getx())/d ey2 = (self.j1.gety()-self.j2.gety())/d px2 = (ex2*(d-self.l)/10000*self.strength) py2 = (ey2*(d-self.l)/10000*self.strength) self.j2.addacc(px2,py2) #izmot rajzoló függvény (most nem szükséges) #def draw(self): #canvas.delete(self.shape) #self.shape=canvas.create_line(self.j1.getx(),self.j1.gety(),self.j2.getx(),self.j2.gety(), fill="indianred") #teljes szimulációs környezetet leíró osztály class Simulation: #bábu definiálása osztályváltozók és egyszerű paraméterek segítségével def __init__(self): self.joint1 = Joint(150,YY-60,0,0,10) self.joint2 = Joint(160,YY-160,0,0,10) self.joint3 = Joint(200,YY-260,0,0,15) self.joint4 = Joint(240,YY-160,0,0,10) self.joint5 = Joint(250,YY-60,0,0,10) self.bone1 = Bone(self.joint1, self.joint2) self.bone2 = Bone(self.joint2, self.joint3) self.bone3 = Bone(self.joint3, self.joint4) self.bone4 = Bone(self.joint4, self.joint5) self.muscle1 = Muscle(self.joint1, self.joint3, 30) self.muscle2 = Muscle(self.joint2, self.joint4, 30) self.muscle3 = Muscle(self.joint3, self.joint5, 30) self.desired_y = 500 self.desired_x = 200 self.aviable_energy = 500000 self.sum_energy = 0 self.reward = 0 self.is_wrong = True self.logfile = open("log0.txt", "w") #reinicializálás def reset(self): self.joint1 = Joint(150,YY-60,0,0,10) self.joint2 = Joint(160,YY-160,0,0,10) self.joint3 = Joint(200,YY-260,0,0,15) self.joint4 = Joint(240,YY-160,0,0,10) self.joint5 = Joint(250,YY-60,0,0,10) self.bone1 = Bone(self.joint1, self.joint2) self.bone2 = Bone(self.joint2, self.joint3) self.bone3 = Bone(self.joint3, self.joint4) self.bone4 = Bone(self.joint4, self.joint5) self.muscle1 = Muscle(self.joint1, self.joint3, 30) self.muscle2 = Muscle(self.joint2, self.joint4, 30) self.muscle3 = Muscle(self.joint3, self.joint5, 30) self.desired_y = YY-260 self.desired_x = 200 self.aviable_energy = 500000 self.sum_energy = 0 self.reward = 0 self.is_wrong = True state_information = [self.joint1.x, self.joint1.y, self.joint2.x, self.joint2.y, self.joint3.x, self.joint3.y, self.joint4.x, self.joint4.y, self.joint5.x, self.joint5.y] return(state_information) #pillanatnyi állapot kiírása fájlba def save_log(self): self.logfile.write(str(self.joint1.getx())) self.logfile.write("\t") self.logfile.write(str(self.joint1.gety())) self.logfile.write("\t") self.logfile.write(str(self.joint2.getx())) self.logfile.write("\t") self.logfile.write(str(self.joint2.gety())) self.logfile.write("\t") self.logfile.write(str(self.joint3.getx())) self.logfile.write("\t") self.logfile.write(str(self.joint3.gety())) self.logfile.write("\t") self.logfile.write(str(self.joint4.getx())) self.logfile.write("\t") self.logfile.write(str(self.joint4.gety())) self.logfile.write("\t") self.logfile.write(str(self.joint5.getx())) self.logfile.write("\t") self.logfile.write(str(self.joint5.gety())) self.logfile.write("\n") #kimeneti fájl megnyitása def open_log(self, num): fname = "log" fname += str(num) fname += ".txt" open(fname, "w").close() self.logfile = open(fname, "w") #kimeneti fájl bezárása def close_log(self): self.logfile.close() #szimuláció leállási paramétereinek ellenőrzése def okay(self): if self.joint3.y>YY-flor-20: return False if self.joint2.y>YY-flor-10: return False if self.joint4.y>YY-flor-10: return False if self.joint1.x>XX or self.joint1.x<0 or self.joint2.x>XX or self.joint2.x<0 or self.joint3.x>XX or self.joint3.x<0 or self.joint4.x>XX or self.joint4.x<0 or self.joint5.x>XX or self.joint5.x<0: return False if self.aviable_energy <=0: return False return True #teljes szimuláció iterálása def step(self, m_prob): #izmok tulajdonságainak változtatása a háló döntése szerint (aktív változtatás) self.muscle1.add_length(5, m_prob[0]) self.muscle1.add_length(-5, m_prob[1]) self.muscle2.add_length(5, m_prob[2]) self.muscle2.add_length(-5, m_prob[3]) self.muscle3.add_length(5, m_prob[4]) self.muscle3.add_length(-5, m_prob[5]) self.muscle1.mult_strength(1.25,m_prob[6]) self.muscle1.mult_strength(0.8,m_prob[7]) self.muscle2.mult_strength(1.25,m_prob[8]) self.muscle2.mult_strength(0.8,m_prob[9]) self.muscle3.mult_strength(1.25,m_prob[10]) self.muscle3.mult_strength(0.8,m_prob[11]) #a bábu állapotának passzív iterálása (belső változóinak függvényében) self.joint1.step() self.joint2.step() self.joint3.step() self.joint4.step() self.joint5.step() self.bone1.repair() self.bone2.repair() self.bone3.repair() self.bone4.repair() self.muscle1.movement() self.muscle2.movement() self.muscle3.movement() #teljes meglévő energia kiszámítása self.sum_energy = self.joint1.energy + self.joint2.energy + self.joint3.energy + self.joint4.energy + self.joint5.energy self.aviable_energy -= self.sum_energy #kimeneti paraméterek meghatározása (háló bemenetek, jutalom függvény, leállást mutató változó) state_information = [self.joint1.x, self.joint1.y, self.joint2.x, self.joint2.y, self.joint3.x, self.joint3.y, self.joint4.x, self.joint4.y, self.joint5.x, self.joint5.y] self.reward = (100000 - pow((self.joint3.y-self.desired_y)/30, 2)- pow((self.joint3.x-self.desired_x)/30, 2))/1000 self.is_wrong = not self.okay() #visszatérés return(state_information, self.reward, self.is_wrong) #szimulációs környezet megvalósítása env = Simulation() env.close_log() #múltbeli állapotok hatásának lecsengése gamma = 0.99 def discount_rewards(r): discounted_r = np.zeros_like(r) running_add = 0 for t in reversed(xrange(0, r.size)): running_add = running_add * gamma + r[t] discounted_r[t] = running_add return discounted_r class agent(): def __init__(self, lr, s_size,a_size,h1_size,h2_size,h3_size): #Neurális háló meghatározása, módosított kimeneti értékekkel együtt self.state_in= tf.placeholder(shape=[None,s_size],dtype=tf.float32) hidden1 = slim.fully_connected(self.state_in,h1_size,biases_initializer=None,activation_fn=tf.nn.relu) hidden2 = slim.fully_connected(hidden1,h2_size,biases_initializer=None,activation_fn=tf.nn.relu) hidden3 = slim.fully_connected(hidden2,h3_size,biases_initializer=None,activation_fn=tf.nn.relu) self.output = slim.fully_connected(hidden3,a_size,activation_fn=tf.nn.sigmoid,biases_initializer=None) outputpairs = tf.reshape(self.output,[-1,6,2]) outputpairs= tf.nn.softmax(outputpairs) #Tanítási folyamat meghatározása a veszteség függvény meghatározásának segítségével self.reward_holder = tf.placeholder(shape=[None],dtype=tf.float32) self.action_holder = tf.placeholder(shape=[None, None],dtype=tf.int32) rewardperaction=tf.tile(tf.expand_dims(self.reward_holder,-1),[1,a_size]) rewardperaction = tf.reshape(rewardperaction,[-1,6,2]) self.loss = -tf.reduce_mean(tf.log(outputpairs)*rewardperaction) #háló egy iterációs lépésnyi tanítása tvars = tf.trainable_variables() self.gradient_holders = [] for idx,var in enumerate(tvars): placeholder = tf.placeholder(tf.float32,name=str(idx)+'_holder') self.gradient_holders.append(placeholder) self.gradients = tf.gradients(self.loss,tvars) optimizer = tf.train.AdamOptimizer(learning_rate=lr) self.update_batch = optimizer.apply_gradients(zip(self.gradient_holders,tvars)) #háló inicializálása tf.reset_default_graph() #háló, és tanítási paramétereinek meghatározása myAgent = agent(lr=1e-2,s_size=10,a_size=12,h1_size=30,h2_size=70,h3_size=30) #maximális tanítási epizódok számának meghatározása total_episodes = 5000 max_ep = 999 update_frequency = 5 init = tf.global_variables_initializer() #Háló tanításának elindítása with tf.Session() as sess: sess.run(init) i = 0 total_reward = [] total_lenght = [] gradBuffer = sess.run(tf.trainable_variables()) for ix,grad in enumerate(gradBuffer): gradBuffer[ix] = grad * 0 while i < total_episodes: s = env.reset() running_reward = 0 ep_history = [] if i % 100 == 0: env.open_log(i/100) for j in range(max_ep): #háló kimenetének meghatározása a = sess.run(myAgent.output,feed_dict={myAgent.state_in:[s]}) a = np.squeeze(a) #szimulációs környezet léptetése egy iterációs lépéssel, majd jutalom meghatározása s1,r,d = env.step(a) ep_history.append([s,a,r,s1]) #minden 100-adik iterációs lépésben szimuláció viselkedésének rögzítése, grafikus megjelenítésre alkalmas fájlként if i % 100 == 0: env.save_log() s = s1 running_reward += r if d == True: #Háló állapotának frissítése ep_history = np.array(ep_history) ep_history[:,2] = discount_rewards(ep_history[:,2]) a=ep_history[:,2] b=np.vstack(ep_history[:,1]) c=np.vstack(ep_history[:,0]) feed_dict={myAgent.reward_holder:a, myAgent.action_holder:b,myAgent.state_in:c} grads = sess.run(myAgent.gradients, feed_dict=feed_dict) for idx,grad in enumerate(grads): gradBuffer[idx] += grad if i % update_frequency == 0 and i != 0: feed_dict= dictionary = dict(zip(myAgent.gradient_holders, gradBuffer)) _ = sess.run(myAgent.update_batch, feed_dict=feed_dict) for ix,grad in enumerate(gradBuffer): gradBuffer[ix] = grad * 0 total_reward.append(running_reward) total_lenght.append(j) break #jutalom értékek kiírása minden 5-ödik iterációs lépésben if i % 5 == 0: print(np.mean(total_reward[-5:])) env.close_log() i += 1