|
|
#!/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(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 ========================
|
|
|
|
|
|
def loaddata():
|
|
|
print('Loading Experiments...')
|
|
|
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.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 to group {}_{} - {} bytes'.format(exp_code, plot, len(str(dum))))
|
|
|
|
|
|
receiver.send_string('ok')
|
|
|
|
|
|
receiver.close()
|
|
|
context.term()
|
|
|
|
|
|
if __name__=='__main__':
|
|
|
loaddata()
|
|
|
main()
|
|
|
|
