ABS/abscontrol Commit - r122:123 · Jicamarca Repository

imanay -

r122:123

parent child

Context file:

r122:123

trunk/absroot/source/abspy/test/threading/file.py created 10644 +2 0

			@@ -0,0 +1,2
		1	import sys
		2	sys.path.insert(0,'/home/redes/workspace/absroot/source/abspy/abscontrol') No newline at end of file

trunk/absroot/source/abspy/test/threading/test1.py created 10644 +22 0

			@@ -0,0 +1,22
		1	import Queue
		2	import threading
		3
		4	def basic_worker(queue):
		5	while True:
		6	item = queue.get()
		7	# do_work(item)
		8	print(item)
		9	queue.task_done()
		10	def basic():
		11	# http://docs.python.org/library/queue.html
		12	queue = Queue.Queue()
		13	for i in range(3):
		14	t = threading.Thread(target=basic_worker,args=(queue,))
		15	t.daemon = True
		16	t.start()
		17	for item in range(4):
		18	queue.put(item)
		19	queue.join() # block until all tasks are done
		20	print('got here')
		21
		22	basic() No newline at end of file

trunk/absroot/source/abspy/test/threading/test2.py created 10644 +33 0

			@@ -0,0 +1,33
		1	import threading
		2	import time
		3	import file
		4	from client3 import *
		5	from server3 import *
		6
		7	def f():
		8	print("get_samples")
		9	cmd = "ANST"
		10	datarpta2 = absObj.getControlModuleStatus(cmd = cmd)
		11
		12	fobj = open("monitoreo.txt","a")
		13	fobj.write(datarpta2)
		14	fobj.close()
		15	threading.Timer(30, f).start()
		16
		17	if __name__ == '__main__':
		18
		19	# absObj = ABSServer(ipDestino="10.10.10.97",rx_buffer = "1") # Adding the same parameter to all the commands
		20	absObj = ABSServer(ipDestino="192.168.1.117",rx_buffer = "1") # Adding the same parameter to all the commands
		21	cmd = "NTST"
		22	datarpta1 = absObj.getConnectionStatus(cmd = cmd)
		23
		24	fobj = open("monitoreo.txt","w")
		25	fobj.write("Monitoring\n")
		26	fobj.write("----------\n")
		27	fobj.close()
		28
		29	f()
		30	while 1:
		31	pass
		32
		33	No newline at end of file

trunk/absroot/source/abspy/test/threading/test3.py created 10644 +8 0

			@@ -0,0 +1,8
		1	import threading as test
		2	import time
		3
		4	def hello():
		5	print "hello, world"
		6
		7	t = test.Timer(1.0, hello)
		8	t.start() # after 30 seconds, "hello, world" will be printed No newline at end of file

General Comments 0

You need to be logged in to leave comments. Login now

No TODOs yet

	Repositories
g s	Goto summary page
g c	Goto changelog page
g f	Goto files page
g F	Goto files page with file search activated
g p	Goto pull requests page
g o	Goto repository settings
g O	Goto repository access permissions settings
t s	Toggle sidebar on some pages