''' @author: Juan C. Espinoza ''' import time import json import numpy import paho.mqtt.client as mqtt import zmq import cPickle as pickle import datetime from zmq.utils.monitor import recv_monitor_message from functools import wraps from threading import Thread from multiprocessing import Process from schainpy.model.proc.jroproc_base import Operation, ProcessingUnit throttle_value = 5 class PrettyFloat(float): def __repr__(self): return '%.2f' % self def roundFloats(obj): if isinstance(obj, list): return map(roundFloats, obj) elif isinstance(obj, float): return round(obj, 2) def pretty_floats(obj): if isinstance(obj, float): return PrettyFloat(obj) elif isinstance(obj, dict): return dict((k, pretty_floats(v)) for k, v in obj.items()) elif isinstance(obj, (list, tuple)): return map(pretty_floats, obj) return obj class throttle(object): """Decorator that prevents a function from being called more than once every time period. To create a function that cannot be called more than once a minute, but will sleep until it can be called: @throttle(minutes=1) def foo(): pass for i in range(10): foo() print "This function has run %s times." % i """ def __init__(self, seconds=0, minutes=0, hours=0): self.throttle_period = datetime.timedelta( seconds=seconds, minutes=minutes, hours=hours ) self.time_of_last_call = datetime.datetime.min def __call__(self, fn): @wraps(fn) def wrapper(*args, **kwargs): now = datetime.datetime.now() time_since_last_call = now - self.time_of_last_call time_left = self.throttle_period - time_since_last_call if time_left > datetime.timedelta(seconds=0): return self.time_of_last_call = datetime.datetime.now() return fn(*args, **kwargs) return wrapper class PublishData(Operation): """Clase publish.""" __MAXNUMX = 100 __MAXNUMY = 100 def __init__(self, **kwargs): """Inicio.""" Operation.__init__(self, **kwargs) self.isConfig = False self.client = None self.zeromq = None self.mqtt = None def on_disconnect(self, client, userdata, rc): if rc != 0: print("Unexpected disconnection.") self.connect() def connect(self): print 'trying to connect' try: self.client.connect( host=self.host, port=self.port, keepalive=60*10, bind_address='') print "connected" self.client.loop_start() # self.client.publish( # self.topic + 'SETUP', # json.dumps(setup), # retain=True # ) except: print "MQTT Conection error." self.client = False def setup(self, port=1883, username=None, password=None, clientId="user", zeromq=1, **kwargs): self.counter = 0 self.topic = kwargs.get('topic', 'schain') self.delay = kwargs.get('delay', 0) self.plottype = kwargs.get('plottype', 'spectra') self.host = kwargs.get('host', "10.10.10.82") self.port = kwargs.get('port', 3000) self.clientId = clientId self.cnt = 0 self.zeromq = zeromq self.mqtt = kwargs.get('plottype', 0) self.client = None setup = [] if mqtt is 1: print 'mqqt es 1' self.client = mqtt.Client( client_id=self.clientId + self.topic + 'SCHAIN', clean_session=True) self.client.on_disconnect = self.on_disconnect self.connect() for plot in self.plottype: setup.append({ 'plot': plot, 'topic': self.topic + plot, 'title': getattr(self, plot + '_' + 'title', False), 'xlabel': getattr(self, plot + '_' + 'xlabel', False), 'ylabel': getattr(self, plot + '_' + 'ylabel', False), 'xrange': getattr(self, plot + '_' + 'xrange', False), 'yrange': getattr(self, plot + '_' + 'yrange', False), 'zrange': getattr(self, plot + '_' + 'zrange', False), }) if zeromq is 1: context = zmq.Context() self.zmq_socket = context.socket(zmq.PUSH) server = kwargs.get('server', 'zmq.pipe') if 'tcp://' in server: address = server else: address = 'ipc:///tmp/%s' % server self.zmq_socket.connect(address) time.sleep(1) print 'zeromq configured' def publish_data(self): self.dataOut.finished = False if self.mqtt is 1: yData = self.dataOut.heightList[:2].tolist() if self.plottype == 'spectra': data = getattr(self.dataOut, 'data_spc') z = data/self.dataOut.normFactor zdB = 10*numpy.log10(z) xlen, ylen = zdB[0].shape dx = numpy.floor(xlen/self.__MAXNUMX) + 1 dy = numpy.floor(ylen/self.__MAXNUMY) + 1 Z = [0 for i in self.dataOut.channelList] for i in self.dataOut.channelList: Z[i] = zdB[i][::dx, ::dy].tolist() payload = { 'timestamp': self.dataOut.utctime, 'data': roundFloats(Z), 'channels': ['Ch %s' % ch for ch in self.dataOut.channelList], 'interval': self.dataOut.getTimeInterval(), 'type': self.plottype, 'yData': yData } # print payload elif self.plottype in ('rti', 'power'): data = getattr(self.dataOut, 'data_spc') z = data/self.dataOut.normFactor avg = numpy.average(z, axis=1) avgdB = 10*numpy.log10(avg) xlen, ylen = z[0].shape dy = numpy.floor(ylen/self.__MAXNUMY) + 1 AVG = [0 for i in self.dataOut.channelList] for i in self.dataOut.channelList: AVG[i] = avgdB[i][::dy].tolist() payload = { 'timestamp': self.dataOut.utctime, 'data': roundFloats(AVG), 'channels': ['Ch %s' % ch for ch in self.dataOut.channelList], 'interval': self.dataOut.getTimeInterval(), 'type': self.plottype, 'yData': yData } elif self.plottype == 'noise': noise = self.dataOut.getNoise()/self.dataOut.normFactor noisedB = 10*numpy.log10(noise) payload = { 'timestamp': self.dataOut.utctime, 'data': roundFloats(noisedB.reshape(-1, 1).tolist()), 'channels': ['Ch %s' % ch for ch in self.dataOut.channelList], 'interval': self.dataOut.getTimeInterval(), 'type': self.plottype, 'yData': yData } elif self.plottype == 'snr': data = getattr(self.dataOut, 'data_SNR') avgdB = 10*numpy.log10(data) ylen = data[0].size dy = numpy.floor(ylen/self.__MAXNUMY) + 1 AVG = [0 for i in self.dataOut.channelList] for i in self.dataOut.channelList: AVG[i] = avgdB[i][::dy].tolist() payload = { 'timestamp': self.dataOut.utctime, 'data': roundFloats(AVG), 'channels': ['Ch %s' % ch for ch in self.dataOut.channelList], 'type': self.plottype, 'yData': yData } else: print "Tipo de grafico invalido" payload = { 'data': 'None', 'timestamp': 'None', 'type': None } # print 'Publishing data to {}'.format(self.host) self.client.publish(self.topic + self.plottype, json.dumps(payload), qos=0) if self.zeromq is 1: print '[Sending] {} - {}'.format(self.dataOut.type, self.dataOut.datatime) self.zmq_socket.send_pyobj(self.dataOut) def run(self, dataOut, **kwargs): self.dataOut = dataOut if not self.isConfig: self.setup(**kwargs) self.isConfig = True self.publish_data() time.sleep(self.delay) def close(self): if self.zeromq is 1: self.dataOut.finished = True self.zmq_socket.send_pyobj(self.dataOut) if self.client: self.client.loop_stop() self.client.disconnect()