realtime_web Commit - r17:9875b8044b93

Better test client & server

Juan C. Espinoza -

r17:9875b8044b93

parent child

Context file:

r17:9875b8044b93

Collapse all files

scripts/client.py created 644 +101 0

@@ -0,0 +1,101
	1	#!/usr/bin/python
	2	# -- coding: UTF-8 --
	3
	4	import time
	5	from datetime import datetime, timedelta
	6	import zmq
	7	import json
	8	import numpy as np
	9	from threading import Thread
	10	import argparse
	11
	12	REQUEST_TIMEOUT = 15000
	13	REQUEST_RETRIES = 10
	14	SERVER_ENDPOINT = 'tcp://localhost:4444'
	15
	16	def send(context, socket, poll, dato):
	17
	18	retries_left = REQUEST_RETRIES
	19
	20	while True:
	21	socket.send_json(dato)
	22	socks = dict(poll.poll(REQUEST_TIMEOUT))
	23	if socks.get(socket) == zmq.POLLIN:
	24	reply = socket.recv_string()
	25	if not reply:
	26	break
	27	if reply == 'ok':
	28	print('I: Server replied OK (%s)' % reply)
	29	break
	30	else:
	31	print('E: Malformed reply from server: %s' % reply)
	32	break
	33	else:
	34	print('W: No response from server, retry {}'.format(REQUEST_RETRIES-retries_left))
	35	time.sleep(2)
	36	socket.setsockopt(zmq.LINGER, 0)
	37	socket.close()
	38	poll.unregister(socket)
	39	retries_left -= 1
	40	if retries_left == 0:
	41	print('E: Server seems to be offline, abandoning')
	42	break
	43	# Create new connection
	44	socket = context.socket(zmq.REQ)
	45	socket.connect(SERVER_ENDPOINT)
	46	poll.register(socket, zmq.POLLIN)
	47
	48	def main(realtime, code, date=None, interval=30):
	49
	50	context = zmq.Context()
	51	socket = context.socket(zmq.REQ)
	52	socket.connect (SERVER_ENDPOINT)
	53	poll = zmq.Poller()
	54	poll.register(socket, zmq.POLLIN)
	55
	56	if realtime:
	57	dt = datetime.now()
	58	else:
	59	dt = date
	60
	61	while True:
	62
	63	print('Sending {} - {}'.format(code, dt))
	64
	65	dato = {
	66	'time': time.mktime(dt.timetuple()),
	67	'yrange': np.arange(100).tolist(),
	68	'xrange': np.arange(-30, 30).tolist(),
	69	'localtime': True,
	70	'interval': interval,
	71	'exp_code': code,
	72	'data': {
	73	'noise': np.round(np.random.rand(8) + np.array([10,11,12,13,14,15,16,17]), 2).tolist(),
	74	'rti': np.round(np.random.rand(8, 100)*5 + 10, 2).tolist(),
	75	'spc': np.round(np.random.rand(8, 60, 100)*5 + 10, 2).tolist(),
	76	}
	77	}
	78
	79	dt = dt + timedelta(seconds=interval)
	80
	81	t = Thread(target=send, args=(context, socket, poll, dato))
	82	t.start()
	83	if realtime:
	84	time.sleep(interval)
	85	else:
	86	time.sleep(5)
	87
	88	if __name__=='__main__':
	89	parser = argparse.ArgumentParser(description='This is a Client for realtime app')
	90	parser.add_argument('--date', action='store', default=None, help='format: 2018/02/13 12:23:00')
	91	parser.add_argument('-r', action='store_true', dest='realtime', default=None)
	92	parser.add_argument('-c', action='store', dest='code', default='170')
	93	parser.add_argument('-i', action='store', dest='interval', type=int, default=30)
	94
	95	results = parser.parse_args()
	96	if not results.realtime:
	97	try:
	98	results.date = datetime.strptime(results.date, '%Y/%m/%d %H:%M:%S')
	99	except:
	100	raise(NameError('You must specify a date (--date) for non-realtime experiment'))
	101	main(results.realtime, results.code, results.date, results.interval)

scripts/server.py +10 -13

             #!/usr/bin/python
             # -*- coding: UTF-8 -*-
             import os
             import sys
             import json
             import simplejson
             from datetime import datetime
             import time
             import zmq
             import redis
             import asgi_redis
             import mongoengine
             from threading import Thread
-            sys.path.append('../')
+            sys.path.append(os.environ.get('APP_DIR', '../'))
             os.environ.setdefault("DJANGO_SETTINGS_MODULE", "realtime.settings")
             from plotter.models import Experiment, ExpMeta, ExpData
             host_mongo = os.environ.get('HOST_MONGO', 'localhost')
             mongoengine.connect('dbplots', host=host_mongo, port=27017)
             host_redis = os.environ.get('HOST_REDIS', 'localhost')
             channel = asgi_redis.RedisChannelLayer(hosts=[(host_redis, 6379)])
             #========================== Conf. ZeroMQ ===================
             context = zmq.Context()
             receiver = context.socket(zmq.REP)
             receiver.bind("tcp://0.0.0.0:4444")
             #============================== end ========================
-            TIMES = {}
             def loaddata():
                 print('Loading Experiments...')
-                for tup in json.load(open('./experiments.json')):
+                if os.environ.get('APP_DIR', None) is not None:
+                    file_exp = os.path.join(os.environ.get('APP_DIR'), 'scripts', 'experiments.json')
+                else:
+                    file_exp = './experiments.json'
+                for tup in json.load(open(file_exp)):
                     print(tup['name'])
                     exp = Experiment.objects(code=tup['code']).modify(
                         upsert=True, # To add a new row
                         new=True,
                         set__code=tup['code'],
                         set__name=tup['name'],
                         )
                     exp.save()
             #============== funcion para modificar datos en la tabla ==============
             def update(buffer):
                 dt = datetime.utcfromtimestamp(buffer['time'])
                 exp = ExpMeta.objects(code=buffer['exp_code'], date=dt.date()).modify(
                     upsert=True, # To add a new row
                     new=True,
                     set__code=buffer['exp_code'],
                     set__date=dt.date(),
                     set__yrange = buffer['yrange'],
                     set__xrange = buffer['xrange'],
                     set__interval = buffer['interval'],
                     set__localtime = buffer['localtime'],
                     set__plots = buffer['data'].keys()
                     )
                 exp.save()
                 data = ExpData.objects(expmeta=exp, time=buffer['time']).modify(
                     upsert=True, # To add a new row
                     new=True,
                     set__expmeta = exp,
                     set__time = buffer['time'],
                     set__data = buffer['data']
                 )
                 data.save()
                 return exp.id
             # Function that is checking the state of my clients every 20s
             def check_times():
                 while True:
-                    dt = datetime.now()
+                    dt = datetime.utcnow()
                     exps = ExpMeta.objects(date=dt.date())
                     for exp in exps:
                         data = ExpData.objects(expmeta=exp).order_by('-time')[0] #me quedo con el ultimo time del ultimo valor
                         t = time.mktime(dt.timetuple())
                         if exp['localtime'] == True: #Consulto que tipode time me esta llegando: LT o UTC
                             t -= 5*60*60
                         if exp['localtime'] == True: #Consulto que tipode time me esta llegando: LT o UTC
                             data_time = data['time'] + 5*60*60
                         if (t-data['time']) > 6*exp['interval']:
                             channel.send_group(u'main', {'text': json.dumps({'code': exp['code'], 'value': 'Offline', 'time': data_time})})
                             print ('{} {} {} {} {}'.format(exp.code, t, data['time'], (t-data['time']), 'offline'))
                         elif (t-data['time']) > 3*exp['interval']:
                             channel.send_group(u'main', {'text': json.dumps({'code': exp['code'], 'value': 'Delayed', 'time': data_time})})
                             print ('{} {} {} {} {}'.format(exp.code, t, data['time'], (t-data['time']), 'delayed'))
                         else:
                             channel.send_group(u'main', {'text': json.dumps({'code': exp['code'], 'value': 'Online', 'time': data_time})})
                             print ('{} {} {} {} {}'.format(exp.code, t, data['time'], (t-data['time']), 'online'))
                     time.sleep(20)
             def main():
                 print('Starting ZMQ server...')
                 t = Thread(target=check_times)
                 t.start()
                 while True:
                     buffer = receiver.recv_json()
                     if buffer['localtime'] == True: # Ask which type of time is coming: LT o UTC
                         buffer['time'] -= 5*60*60
                     if not isinstance(buffer, dict):
                         print('*******************')
                         print(buffer)
                         continue
                     code = update(buffer)
                     print("==================================")
                     for plot in buffer['data']:
                         dum = buffer.copy()
                         dum['time'] = [buffer['time']]
                         if plot=='noise':
                             dum[plot] = [[x] for x in buffer['data'][plot]]
                         elif plot=='spc':
                             dum['noise'] = [[x] for x in buffer['data']['noise']]
                             dum['spc'] = buffer['data']['spc']
                             dum['rti'] = buffer['data']['rti']
                         else:
                             dum[plot] = buffer['data'][plot]
                         dum.pop('data')
                         exp_code = dum.pop('exp_code')
                         #code = dum.pop('exp_code')
                         channel.send_group(u'{}_{}'.format(exp_code, plot), {'text': simplejson.dumps(dum, ignore_nan=True)})
-                        print('Sending...{}:{} - {} bytes'.format(exp_code, plot, len(str(dum))))
+                        print('Sending to group {}_{} - {} bytes'.format(exp_code, plot, len(str(dum))))
-                    receiver.send_string('recibido')
+                    receiver.send_string('ok')
                 receiver.close()
                 context.term()
             if __name__=='__main__':
                 loaddata()
                 main()

scripts/client3.py removed 0 -124

NO CONTENT: file was removed

scripts/client4.py removed 0 -122

NO CONTENT: file was removed

General Comments 0

Write
Preview

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

No TODOs yet

	Site-wide shortcuts
/	Use quick search box
g h	Goto home page
g g	Goto my private gists page
g G	Goto my public gists page
g 0-9	Goto bookmarked items from 0-9
n r	New repository page
n g	New gist page

	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