diff --git a/scripts/client.py b/scripts/client.py new file mode 100644 index 0000000..e1a5257 --- /dev/null +++ b/scripts/client.py @@ -0,0 +1,101 @@ +#!/usr/bin/python +# -*- coding: UTF-8 -*- + +import time +from datetime import datetime, timedelta +import zmq +import json +import numpy as np +from threading import Thread +import argparse + +REQUEST_TIMEOUT = 15000 +REQUEST_RETRIES = 10 +SERVER_ENDPOINT = 'tcp://localhost:4444' + +def send(context, socket, poll, dato): + + retries_left = REQUEST_RETRIES + + while True: + socket.send_json(dato) + socks = dict(poll.poll(REQUEST_TIMEOUT)) + if socks.get(socket) == zmq.POLLIN: + reply = socket.recv_string() + if not reply: + break + if reply == 'ok': + print('I: Server replied OK (%s)' % reply) + break + else: + print('E: Malformed reply from server: %s' % reply) + break + else: + print('W: No response from server, retry {}'.format(REQUEST_RETRIES-retries_left)) + time.sleep(2) + socket.setsockopt(zmq.LINGER, 0) + socket.close() + poll.unregister(socket) + retries_left -= 1 + if retries_left == 0: + print('E: Server seems to be offline, abandoning') + break + # Create new connection + socket = context.socket(zmq.REQ) + socket.connect(SERVER_ENDPOINT) + poll.register(socket, zmq.POLLIN) + +def main(realtime, code, date=None, interval=30): + + context = zmq.Context() + socket = context.socket(zmq.REQ) + socket.connect (SERVER_ENDPOINT) + poll = zmq.Poller() + poll.register(socket, zmq.POLLIN) + + if realtime: + dt = datetime.now() + else: + dt = date + + while True: + + print('Sending {} - {}'.format(code, dt)) + + dato = { + 'time': time.mktime(dt.timetuple()), + 'yrange': np.arange(100).tolist(), + 'xrange': np.arange(-30, 30).tolist(), + 'localtime': True, + 'interval': interval, + 'exp_code': code, + 'data': { + 'noise': np.round(np.random.rand(8) + np.array([10,11,12,13,14,15,16,17]), 2).tolist(), + 'rti': np.round(np.random.rand(8, 100)*5 + 10, 2).tolist(), + 'spc': np.round(np.random.rand(8, 60, 100)*5 + 10, 2).tolist(), + } + } + + dt = dt + timedelta(seconds=interval) + + t = Thread(target=send, args=(context, socket, poll, dato)) + t.start() + if realtime: + time.sleep(interval) + else: + time.sleep(5) + +if __name__=='__main__': + parser = argparse.ArgumentParser(description='This is a Client for realtime app') + parser.add_argument('--date', action='store', default=None, help='format: 2018/02/13 12:23:00') + parser.add_argument('-r', action='store_true', dest='realtime', default=None) + parser.add_argument('-c', action='store', dest='code', default='170') + parser.add_argument('-i', action='store', dest='interval', type=int, default=30) + + results = parser.parse_args() + if not results.realtime: + try: + results.date = datetime.strptime(results.date, '%Y/%m/%d %H:%M:%S') + except: + raise(NameError('You must specify a date (--date) for non-realtime experiment')) + main(results.realtime, results.code, results.date, results.interval) diff --git a/scripts/client3.py b/scripts/client3.py deleted file mode 100644 index 789fff9..0000000 --- a/scripts/client3.py +++ /dev/null @@ -1,124 +0,0 @@ -#!/usr/bin/python -# -*- coding: UTF-8 -*- - -import time -from datetime import datetime, timedelta -import zmq -import json -import random - - -REQUEST_TIMEOUT = 15000 -REQUEST_RETRIES = 10 -SERVER_ENDPOINT = "tcp://localhost:4444" - - -context = zmq.Context() - -#print("Connecting to server…") -socket = context.socket(zmq.REQ) -socket.connect (SERVER_ENDPOINT) - -print('Starting ZMQ client...') -poll = zmq.Poller() -poll.register(socket, zmq.POLLIN) - -sequence = 0 -retries_left = REQUEST_RETRIES - -#dt=datetime(2018,5,15) #en vez de poner now() -dt=datetime.now() -while retries_left: - sequence += 1 - request = str(sequence).encode() - print("HF_JRO: Sending (%s)" % request) - - #==== datos noise ========= - n1 = random.uniform(20.0,20.2) - n2 = random.uniform(21.0,21.3) - n3 = random.uniform(22.0,22.2) - n4 = random.uniform(23.0,23.2) - n5 = random.uniform(24.0,24.2) - n6 = random.uniform(25.0,25.3) - n7 = random.uniform(26.0,26.2) - n8 = random.uniform(27.0,27.2) - #=== fin de datos noise === - - #==== datos rti =========== - yrange = list(range(0,200,2)) - elementos=[] - while len(elementos)<=100: - elementos.append(random.randint(15, 35)) - #=== fin de datos rti ===== - - - #==== datos spc =========== - x_range = list(range(80)) - matrix=[] - while len(matrix)<=80: - matrix.append(elementos) - #=== fin de datos spc ===== - - dt1=dt+timedelta(seconds=30) #ahora voy a cambiar cada 30min ya no cada 30 segundos - #dt1 = datetime.now() - - dato1 = {"time":time.mktime(dt1.timetuple()), - "yrange": yrange, - "xrange": x_range, - "localtime": True, - "interval": 30, - "exp_code": 100, - "data": {"noise":[n1, n2, n3, n4, n5, n6, n7, n8], - "rti":[elementos,elementos,elementos,elementos,elementos,elementos,elementos,elementos], - "spc":[matrix,matrix,matrix,matrix,matrix,matrix,matrix,matrix] - } - } - - #print("Sending..") - dt = dt1 - - #=== para simular huecos === - #if dt.hour in (3 ,4): - # continue - socket.send_json(dato1) - expect_reply = True - while expect_reply: - socks = dict(poll.poll(REQUEST_TIMEOUT)) - if socks.get(socket) == zmq.POLLIN: - reply = socket.recv_string() - if not reply: - break - if reply == "recibido": - print("I: Server replied OK (%s)" % reply) - retries_left = REQUEST_RETRIES - expect_reply = False - else: - print("E: Malformed reply from server: %s" % reply) - - else: - print("W: No response from server, retrying…") - # Socket is confused. Close and remove it. - socket.setsockopt(zmq.LINGER, 0) - socket.close() - poll.unregister(socket) - retries_left -= 1 - if retries_left == 0: - print("E: Server seems to be offline, abandoning") - break - print("I: Reconnecting and resending (%s)" % request) - # Create new connection - socket = context.socket(zmq.REQ) - socket.connect(SERVER_ENDPOINT) - poll.register(socket, zmq.POLLIN) - #socket.send(request) - #time.sleep(5) - time.sleep(30) -context.term() - - - - - -#=================== - - diff --git a/scripts/client4.py b/scripts/client4.py deleted file mode 100644 index 14c791d..0000000 --- a/scripts/client4.py +++ /dev/null @@ -1,122 +0,0 @@ -#!/usr/bin/python -# -*- coding: UTF-8 -*- - -import time -from datetime import datetime, timedelta -import zmq -import json -import random - - -REQUEST_TIMEOUT = 15000 -REQUEST_RETRIES = 10 -SERVER_ENDPOINT = "tcp://localhost:4444" - - -context = zmq.Context() - -#print("Connecting to server…") -socket = context.socket(zmq.REQ) -socket.connect (SERVER_ENDPOINT) - -print('Starting ZMQ client...') -poll = zmq.Poller() -poll.register(socket, zmq.POLLIN) - -sequence = 0 -retries_left = REQUEST_RETRIES - -#dt=datetime(2018,5,15) #en vez de poner now() -dt=datetime.now() -while retries_left: - sequence += 1 - request = str(sequence).encode() - print("HF_JRO: Sending (%s)" % request) - - #==== datos noise ========= - n1 = random.uniform(20.0,20.2) - n2 = random.uniform(21.0,21.3) - n3 = random.uniform(22.0,22.2) - n4 = random.uniform(23.0,23.2) - n5 = random.uniform(24.0,24.2) - n6 = random.uniform(25.0,25.3) - n7 = random.uniform(26.0,26.2) - n8 = random.uniform(27.0,27.2) - #=== fin de datos noise === - - #==== datos rti =========== - yrange = list(range(0,200,2)) - elementos=[] - while len(elementos)<=100: - elementos.append(random.randint(15, 35)) - #=== fin de datos rti ===== - - - #==== datos spc =========== - x_range = list(range(80)) - matrix=[] - while len(matrix)<=80: - matrix.append(elementos) - #=== fin de datos spc ===== - - dt1=dt+timedelta(seconds=30) #ahora voy a cambiar cada 30min ya no cada 30 segundos - #dt1 = datetime.now() - - dato1 = {"time":time.mktime(dt1.timetuple()), - "yrange": yrange, - "xrange": x_range, - "localtime": True, - "interval": 30, - "exp_code": 101, - "data": {"noise":[n1, n2, n3, n4, n5, n6, n7, n8], - "rti":[elementos,elementos,elementos,elementos,elementos,elementos,elementos,elementos], - "spc":[matrix,matrix,matrix,matrix,matrix,matrix,matrix,matrix] - } - } - - #print("Sending..") - dt = dt1 - - #=== para simular huecos === - #if dt.hour in (3 ,4): - # continue - socket.send_json(dato1) - expect_reply = True - while expect_reply: - socks = dict(poll.poll(REQUEST_TIMEOUT)) - if socks.get(socket) == zmq.POLLIN: - reply = socket.recv_string() - if not reply: - break - if reply == "recibido": - print("I: Server replied OK (%s)" % reply) - retries_left = REQUEST_RETRIES - expect_reply = False - else: - print("E: Malformed reply from server: %s" % reply) - - else: - print("W: No response from server, retrying…") - # Socket is confused. Close and remove it. - socket.setsockopt(zmq.LINGER, 0) - socket.close() - poll.unregister(socket) - retries_left -= 1 - if retries_left == 0: - print("E: Server seems to be offline, abandoning") - break - print("I: Reconnecting and resending (%s)" % request) - # Create new connection - socket = context.socket(zmq.REQ) - socket.connect(SERVER_ENDPOINT) - poll.register(socket, zmq.POLLIN) - #socket.send(request) - #time.sleep(5) - time.sleep(30) -context.term() - - - - - -#=================== \ No newline at end of file diff --git a/scripts/server.py b/scripts/server.py index fb47c26..6f16cf3 100644 --- a/scripts/server.py +++ b/scripts/server.py @@ -13,7 +13,7 @@ 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 @@ -21,7 +21,6 @@ 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 =================== @@ -30,11 +29,13 @@ 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 @@ -75,7 +76,7 @@ def update(buffer): def check_times(): while True: - dt = datetime.now() + dt = datetime.utcnow() exps = ExpMeta.objects(date=dt.date()) for exp in exps: @@ -86,8 +87,6 @@ def check_times(): 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 @@ -108,12 +107,10 @@ def main(): t = Thread(target=check_times) t.start() - while True: + 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 @@ -139,9 +136,9 @@ def main(): #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()