How to use psput method in fMBT
Best Python code snippet using fMBT_python
trainCM.py
Source:trainCM.py 
1#!/usr/bin/env python22# -*- coding: utf-8 -*-3"""4Created on Thu Nov 16 11:08:01 20175@author: p6"""7import numpy as np8import matplotlib.pyplot as plt9import tensorflow as tf10import time11from epics import caget,caput12fName='trainCMData'13dataOri=np.loadtxt(fName)14dataBPM=dataOri[:,6:22]15dataPS=dataOri[:,70:86]16numItem,numBPM= np.shape(dataBPM)17numPS=np.shape(dataPS)[1]18##-------------##----------------##-------------------19def GenWeight(shape):20 initial = tf.truncated_normal(shape, stddev=1.)21 return tf.Variable(initial)22def GenBias(shape):23 initial=GenWeight((1,shape(0)))24 return initial[0,:]25 26def PreTrain(psSetRec,bpmRec,batchSize):27 numItem=np.shape(bpmRec)[0]28 idChoose_0=np.random.randint(0,high=numItem,size=(batchSize))29 idChoose_1=np.random.randint(1,high=numItem,size=(batchSize))30 31 d_psChoose=psSetRec[idChoose_1,:]-psSetRec[idChoose_0,:]32 d_BPMChoose=bpmRec[idChoose_1,:]-bpmRec[idChoose_0,:]33 34 # x0=bpmRec[idChoose_0,:]35 #dx=d_BPMChoose36 #X=np.hstack((x0,dx))37 38 X=d_BPMChoose39 Y=d_psChoose40 41 return X,Y42def RunTrain(se,opt,loss,numEpoch,batchSize,psSetRec,bpmRec):43 lossRecRec=[]44 for _ in range(numEpoch):45 X,Y=PreTrain(psSetRec,bpmRec,batchSize)46 47 se.run(opt,feed_dict={xIn:X,yIn:Y})48 if _% 50==0:49 lossRecTmp=se.run(loss,feed_dict={xIn:X,yIn:Y})50 lossRecRec.append(lossRecTmp)51 return lossRecRec52 53learningRate=0.005 #--------------------------------------------54numEpoch=50000055batchSize=5056numInput=numBPM57numOutput=numPS58xIn=tf.placeholder(tf.float32,shape=(None,numInput))59yIn=tf.placeholder(tf.float32,shape=(None,numOutput))60'''61num1=862w1=GenWeight((numInput,num1))63b1= GenWeight((1,num1))[0,:]64x1=tf.nn.relu(tf.nn.xw_plus_b(xIn,w1,b1))65#66num2=numOutput67w2=GenWeight((num1,num2))68b2= GenWeight((1,num2))[0,:]69x2=tf.nn.xw_plus_b(x1,w2,b2)70'''71w0=GenWeight((numInput,numOutput))72x0=tf.matmul(xIn,w0)73xOut=x074yOut=yIn75loss=tf.losses.mean_squared_error(xOut,yOut)76train=tf.train.AdamOptimizer(learningRate)77opt=train.minimize(loss)78if vars().has_key('se'):79 se.close()80else:81 se=tf.Session()82se.run(tf.global_variables_initializer())83######## --------- Train ------------------84plt.close('all')85lossRec=[]86lossMeanRec=[]87rAimRec=[]88for _ in range(numEpoch):89 X,Y=PreTrain(dataPS,dataBPM,batchSize)90 91 se.run(opt,feed_dict={xIn:X,yIn:Y})92 if _% 50==0:93 lossRecTmp=se.run(loss,feed_dict={xIn:X,yIn:Y})94 xAim=np.zeros((1,numBPM))-np.mean(dataBPM,axis=0)[np.newaxis,:]95 yAim=se.run(xOut,feed_dict={xIn:xAim})[0]96 lossRec.append(lossRecTmp)97 lossMeanRec.append(np.mean(lossRec))98 rAimRec.append(np.sum(np.square(yAim)))99 if len(lossRec)>200:100 lossRec.pop(0)101 lossMeanRec.pop(0)102 rAimRec.pop(0)103 plt.figure(1)104 plt.clf()105 plt.hold('on')106 plt.plot(lossRec,'r')107 plt.plot(lossMeanRec,'g')108 plt.grid('on')109 plt.title(_)110 plt.pause(0.01)111 112 plt.figure(2)113 plt.clf()114 plt.plot(yAim,'r-*')115 plt.grid('on')116 plt.title(_)117 plt.pause(0.01)118 119 plt.figure(3)120 plt.clf()121 plt.plot(rAimRec,'g-')122 plt.grid('on')123 plt.title(_)124 plt.pause(0.01) 125 ##-------- Final ------------------------126psList=[\127 'MEBT_PS:DCH_06:ISet','MEBT_PS:DCV_06:ISet', \128 'MEBT_PS:DCH_07:ISet','MEBT_PS:DCV_07:ISet', \129 'HCM1_PS:DCH_01:ISet','HCM1_PS:DCV_01:ISet', \130 'HCM1_PS:DCH_02:ISet','HCM1_PS:DCV_02:ISet', \131 'HCM1_PS:DCH_03:ISet','HCM1_PS:DCV_03:ISet', \132 'HCM1_PS:DCH_04:ISet','HCM1_PS:DCV_04:ISet', \133 'HCM1_PS:DCH_05:ISet','HCM1_PS:DCV_05:ISet', \134 'HCM1_PS:DCH_06:ISet','HCM1_PS:DCV_06:ISet', \135 ]136bpmList=[\137 'BPM:4-X11','BPM:4-Y11',\138 'BPM:5-X11','BPM:5-Y11',\139 'BPM:6-X11', 'BPM:6-Y11',\140 'BPM:7-X11','BPM:7-Y11',\141 'BPM:8-X11','BPM:8-Y11',\142 'BPM:9-X11','BPM:9-Y11',\143 'BPM:10-X11','BPM:10-Y11',\144 'BPM:11-X11','BPM:11-Y11',\145 ]146def GetBPM(bpmList):147 bpmNow=[]148 for iBPM in bpmList:149 bpmNow.append(caget(iBPM))150 bpmNow=np.array(bpmNow)[np.newaxis,:]151 return bpmNow152def GetPS(psSetList):153 psNow=[]154 for iPS in psSetList:155 psNow.append(caget(iPS))156 psNow=np.array(psNow)[np.newaxis,:]157 return psNow158def PutPS(psSetList,psSetNow):159 for i in range(len(psSetList)):160 iStr,iVal=psSetList[i],psSetNow[i]161 caput(iStr,iVal)162def LogWrite(fid,X):163 fid.writelines('#=========================')164 timeNow='# '+time.asctime()165 fid.writelines(timeNow)166 fid.writelines('\n')167 168 for i in X:169 for j in i:170 fid.writelines('%.2f ' %j)171 fid.writelines('\n')172 173bpmNow=GetBPM(bpmList)174bpmAim=0-bpmNow175psAim=se.run(xOut,feed_dict={xIn:bpmAim})176psNow=GetPS(psList)177psPut=(psNow+psAim)[0]178for i in range(len(psAim)):179 if i <4:180 if psPut[i]<-15:181 psPut[i]=-15182 if psPut[i]>15:183 psPut[i]=15184 else:185 if psPut[i]<-65:186 psPut[i]=-65187 if psPut[i]>65:188 psPut[i]=65189psPut=np.round(psPut*100)/100190print psPut191PutPS(psList,psPut)192time.sleep(80)193bpmUpdate=GetBPM(bpmList)194fid=open('TestLogVIP','a+')195LogWrite(fid,(bpmNow[0],bpmAim[0],psAim[0],psNow[0],psPut,bpmUpdate[0]))196fid.close()197plt.figure(9986)198plt.plot(psAim[0],'-*')199plt.figure(9987)200plt.plot(psNow[0],'-*')201plt.figure(9988)202plt.plot(bpmUpdate[0],'-*')203'''204for i in range(numBPM):205 plt.figure('BPM')206 plt.clf()207 plt.hist(dataBPM[:,i],100)208 plt.grid('on')209 plt.title(i)210 plt.pause(1)211for i in range(numPS):212 plt.figure('PS')213 plt.clf()214 plt.hist(dataPS[:,i],100)215 plt.grid('on')216 plt.title(i)217 plt.pause(1)218'''219'''220class NNMat():221 def __init__(self,numInput,numOutput,learningRate):222 self.numInput=numInput223 self.numOutput=numOutput224 self.learningRate=learningRate225 def GenWeight(shape):226 initial = tf.truncated_normal(shape, stddev=1.)227 return tf.Variable(initial)228 def GenBias(shape):229 initial = tf.constant(1., shape=shape)230 return tf.Variable(initial)231'''232 ...
pubsub.py
Source:pubsub.py
1__all__ = ['PublishSubscribeQueue']2import events3from util import priority45class PSPut(events.Operation):6 __slots__ = ('queue', 'message', 'key')78 def __init__(self, queue, message, key, **kws):9 super(PSPut, self).__init__(**kws)10 self.queue = queue11 self.message = message12 self.key = key1314 def process(self, sched, coro):15 super(PSPut, self).process(sched, coro)16 self.queue.messages.append(self.message)17 result = [self.message]18 for getkey in self.queue.active_subscribers:19 self.queue.subscribers[getkey] += 120 getop, getcoro = self.queue.active_subscribers[getkey]21 getop.result = result22 if getop.prio:23 sched.active.appendleft((getop, getcoro))24 else:25 sched.active.append((getop, getcoro))26 self.queue.active_subscribers.clear()27 if self.prio & priority.CORO:28 return self, coro29 else:30 if self.prio & priority.OP:31 sched.active.appendleft((self, coro))32 else:33 sched.active.append((self, coro))34 return None, None3536class PSGet(events.TimedOperation):37 __slots__ = ('queue', 'result', 'key')3839 def __init__(self, queue, key, **kws):40 super(PSGet, self).__init__(**kws)41 self.queue = queue42 self.key = key4344 def process(self, sched, coro):45 super(PSGet, self).process(sched, coro)46 key = self.key or coro47 assert key in self.queue.subscribers48 level = self.queue.subscribers[key]49 queue_level = len(self.queue.messages)50 if level < queue_level:51 self.result = self.queue.messages[level:] if level \52 else self.queue.messages53 self.queue.subscribers[key] = queue_level54 return self, coro55 else:56 self.queue.active_subscribers[key] = self, coro5758 def finalize(self, sched):59 super(PSGet, self).finalize(sched)60 return self.result6162 def cleanup(self, sched, coro):63 for key in self.queue.active_subscribers:64 getop, getcoro = self.queue.active_subscribers[key]65 if coro is getcoro:66 assert getop is self6768 del self.queue.active_subscribers[key]69 return True7071class PSSubscribe(events.Operation):72 __slots__ = ('queue', 'key')7374 def __init__(self, queue, key, **kws):75 super(PSSubscribe, self).__init__(**kws)76 self.queue = queue77 self.key = key7879 def process(self, sched, coro):80 super(PSSubscribe, self).process(sched, coro)81 self.queue.subscribers[self.key or coro] = 082 return self, coro8384class PSUnsubscribe(events.Operation):85 __slots__ = ('queue', 'key')8687 def __init__(self, queue, key, **kws):88 super(PSUnsubscribe, self).__init__(**kws)89 self.queue = queue90 self.key = key9192 def process(self, sched, coro):93 super(PSUnsubscribe, self).process(sched, coro)94 del self.queue.subscribers[self.key or coro]95 return self, coro969798class PublishSubscribeQueue:99 """A more robust replacement for the signal operations.100 A coroutine subscribes itself to a PublishSubscribeQueue and get new101 published messages with _fetch_ method.102 """103 def __init__(self):104 self.messages = []105 self.subscribers = {}106 self.active_subscribers = {} # holds waiting fetch ops107108 def publish(self, message, key=None, **kws):109 """Put a message in the queue and updates any coroutine wating with110 fetch. *works as a coroutine operation*"""111 return PSPut(self, message, key, **kws)112113 def subscribe(self, key=None, **kws):114 """Registers the calling coroutine to the queue. Sets the update index115 to 0 - on fetch, that coroutine will get all the messages from the116 queue. *works as a coroutine operation*"""117 return PSSubscribe(self, key, **kws)118119 def unsubscribe(self, key=None, **kws):120 """Unregisters the calling coroutine to the queue. """121 # TODO: unittest122 return PSUnsubscribe(self, key, **kws)123124 def fetch(self, key=None, **kws):125 """Get all the new messages since the last fetch. Returns a list126 of messages. *works as a coroutine operation*"""127 return PSGet(self, key, **kws)128129 def compact(self):130 """Compacts the queue: removes all the messages from the queue that131 have been fetched by all the subscribed coroutines.132 Returns the number of messages that have been removed."""133 if self.subscribers:134 level = min(self.subscribers.itervalues())135 if level:136 del self.messages[:level]137 return level138 else:139 level = len(self.messages)140 del self.messages[:]
...
task_7_1.py
Source:task_7_1.py
1# задаÑа 7. ÐаÑÐ¸Ð°Ð½Ñ 1.2# ÐапиÑиÑе пÑогÑаммÑ, коÑоÑÐ°Ñ Ð±Ñ Ð¿Ñи запÑÑке ÑлÑÑайнÑм обÑазом оÑобÑажала3# название одного из Ñеми ÑÑÐ´ÐµÑ ÑвеÑаÑ4# ÐолаÑÑкаев.Ð.Ð5# 21.03.20176import random7sput = ["ÐÑоÑ", "ÐоÑÑоÑ", "ÐÑамиÑ"]8rsput = random.choice(sput)9score = 1010psput = ""11while psput != rsput:12 psput = input("ÐазовиÑе одиного из ÑÑеÑ
ÑоваÑиÑей д'ÐÑÑанÑÑна: ")13 if psput == rsput:14 print("\nХоÑоооÑ!")15 print("У ваÑ", score,"баллов.")16 else:17 score-=518 print("\nÐе пÑавилÑно((\nÐопÑобÑйÑе еÑе Ñаз\n\n")...
Automation Testing Tutorials
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run fMBT automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
