schain Commit - r904:1253852f58ac · Jicamarca Repository

receiver data modificado para web

José Chávez -

r904:1253852f58ac

parent child

Context file:

r904:1253852f58ac

Collapse all files

schainpy/model/utils/jroutils_publish.py +10 -5

             '''
             @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
             MAXNUMX = 100
             MAXNUMY = 100
             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 decimate(z):
+                # dx = int(len(self.x)/self.__MAXNUMX) + 1
+                dy = int(len(z[0])/MAXNUMY) + 1
+                return z[::, ::dy]
             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."""
                 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='')
                         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:
                         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)
                 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 = int(xlen/MAXNUMX) + 1
                             dy = int(ylen/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()
             class ReceiverData(ProcessingUnit, Process):
                 throttle_value = 5
                 def __init__(self, **kwargs):
                     ProcessingUnit.__init__(self, **kwargs)
                     Process.__init__(self)
                     self.mp = False
                     self.isConfig = False
                     self.plottypes =[]
                     self.connections = 0
                     server = kwargs.get('server', 'zmq.pipe')
                     if 'tcp://' in server:
                         address = server
                     else:
                         address = 'ipc:///tmp/%s' % server
                     self.address = address
                     self.plottypes = [s.strip() for s in kwargs.get('plottypes', 'rti').split(',')]
                     self.realtime = kwargs.get('realtime', False)
                     self.throttle_value = kwargs.get('throttle', 10)
                     self.sendData = self.initThrottle(self.throttle_value)
                     self.setup()
                 def setup(self):
                     self.data = {}
                     self.data['times'] = []
                     for plottype in self.plottypes:
                         self.data[plottype] = {}
                     self.data['noise'] = {}
                     self.data['throttle'] = self.throttle_value
                     self.data['ENDED'] = False
                     self.isConfig = True
                     self.data_web = {}
                 def event_monitor(self, monitor):
                     events = {}
                     for name in dir(zmq):
                         if name.startswith('EVENT_'):
                             value = getattr(zmq, name)
                             events[value] = name
                     while monitor.poll():
                         evt = recv_monitor_message(monitor)
                         if evt['event'] == 32:
                             self.connections += 1
                         if evt['event'] == 512:
                             pass
                         if self.connections == 0 and self.started is True:
                             self.ended = True
                             # send('ENDED')
                         evt.update({'description': events[evt['event']]})
                         if evt['event'] == zmq.EVENT_MONITOR_STOPPED:
                             break
                     monitor.close()
                     print("event monitor thread done!")
                 def initThrottle(self, throttle_value):
                     @throttle(seconds=throttle_value)
                     def sendDataThrottled(fn_sender, data):
                         fn_sender(data)
                     return sendDataThrottled
                 def send(self, data):
-                    print '[sending] data=%s size=%s' % (data.keys(), len(data['times']))
+                    # print '[sending] data=%s size=%s' % (data.keys(), len(data['times']))
                     self.sender.send_pyobj(data)
                 def update(self):
                     t = self.dataOut.ltctime
                     self.data['times'].append(t)
                     self.data['dataOut'] = self.dataOut
                     for plottype in self.plottypes:
                         if plottype == 'spc':
                             z = self.dataOut.data_spc/self.dataOut.normFactor
                             self.data[plottype] = 10*numpy.log10(z)
                             self.data['noise'][t] = 10*numpy.log10(self.dataOut.getNoise()/self.dataOut.normFactor)
                         if plottype == 'rti':
                             self.data[plottype][t] = self.dataOut.getPower()
                         if plottype == 'snr':
                             self.data[plottype][t] = 10*numpy.log10(self.dataOut.data_SNR)
                         if plottype == 'dop':
                             self.data[plottype][t] = 10*numpy.log10(self.dataOut.data_DOP)
                         if plottype == 'coh':
                             self.data[plottype][t] = self.dataOut.getCoherence()
                         if plottype == 'phase':
                             self.data[plottype][t] = self.dataOut.getCoherence(phase=True)
                         if self.realtime:
-                            self.data_web[plottype] = self.data[plottype][t]
+                            self.data_web[plottype] = roundFloats(decimate(self.data[plottype][t]).tolist())
                             self.data_web['time'] = t
                 def run(self):
                     print '[Starting] {} from {}'.format(self.name, self.address)
                     self.context = zmq.Context()
                     self.receiver = self.context.socket(zmq.PULL)
                     self.receiver.bind(self.address)
                     monitor = self.receiver.get_monitor_socket()
                     self.sender = self.context.socket(zmq.PUB)
                     if self.realtime:
                         self.sender_web = self.context.socket(zmq.PUB)
-                        self.sender.bind("ipc:///tmp/zmq.web")
+                        # self.sender_web.setsockopt(zmq.PUBLISH, 'realtime')
+                        self.sender_web.bind("ipc:///tmp/zmq.web")
                     self.sender.bind("ipc:///tmp/zmq.plots")
-                    t = Thread(target=self.event_monitor)
+                    t = Thread(target=self.event_monitor, args=(monitor,))
                     t.start()
                     while True:
                         self.dataOut = self.receiver.recv_pyobj()
                         # print '[Receiving] {} - {}'.format(self.dataOut.type,
                         #                                    self.dataOut.datatime.ctime())
                         self.update()
                         if self.dataOut.finished is True:
                             self.send(self.data)
                             self.connections -= 1
                             if self.connections == 0 and self.started:
                                 self.ended = True
                                 self.data['ENDED'] = True
                                 self.send(self.data)
                                 self.setup()
                         else:
                             if self.realtime:
                                 self.send(self.data)
-                                self.sender_web.send_json(json.dumps(self.data_web))
+                                self.sender_web.send_string(json.dumps(self.data_web))
                             else:
                                 self.sendData(self.send, self.data)
                             self.started = True
                     return

schainpy/scripts/PPD.py +2 -2

             import argparse
             from schainpy.controller import Project, multiSchain
             desc = "HF_EXAMPLE"
             def fiber(cursor, skip, q, dt):
                 controllerObj = Project()
                 controllerObj.setup(id='191', name='test01', description=desc)
                 readUnitConfObj = controllerObj.addReadUnit(datatype='SpectraReader',
                                                              path='/home/nanosat/data/hysell_data20/pdata',
                                                              startDate=dt,
                                                              endDate=dt,
                                                              startTime="00:00:00",
                                                              endTime="23:59:59",
                                                              online=0,
                                                              #set=1426485881,
                                                              delay=10,
                                                              walk=1,
                                                              queue=q,
                                                              cursor=cursor,
                                                              skip=skip,
                                                              #timezone=-5*3600
                                                              )
                 #     #opObj11 = readUnitConfObj.addOperation(name='printNumberOfBlock')
                 #
                 procUnitConfObj2 = controllerObj.addProcUnit(datatype='Spectra', inputId=readUnitConfObj.getId())
                 # opObj11 = procUnitConfObj2.addParameter(name='pairsList', value='(0,1)', format='pairslist')
                 #
                 # procUnitConfObj3 = controllerObj.addProcUnit(datatype='ParametersProc', inputId=readUnitConfObj.getId())
                 # opObj11 = procUnitConfObj3.addOperation(name='SpectralMoments', optype='other')
                 #
                 #     opObj11 = procUnitConfObj1.addOperation(name='SpectraPlot', optype='other')
                 #     opObj11.addParameter(name='id', value='1000', format='int')
                 #     opObj11.addParameter(name='wintitle', value='HF_Jicamarca_Spc', format='str')
                 #     opObj11.addParameter(name='channelList', value='0', format='intlist')
                 #     opObj11.addParameter(name='zmin', value='-120', format='float')
                 #     opObj11.addParameter(name='zmax', value='-70', format='float')
                 #     opObj11.addParameter(name='save', value='1', format='int')
                 #     opObj11.addParameter(name='figpath', value=figpath, format='str')
                 # opObj11 = procUnitConfObj2.addOperation(name='RTIPlot', optype='other')
                 # opObj11.addParameter(name='id', value='2000', format='int')
                 # opObj11.addParameter(name='wintitzmaxle', value='HF_Jicamarca', format='str')
                 # opObj11.addParameter(name='showprofile', value='0', format='int')
                 # # opObj11.addParameter(name='channelList', value='0', format='intlist')
                 # #     opObj11.addParameter(name='xmin', value='0', format='float')
                 # opObj11.addParameter(name='xmin', value='0', format='float')
                 # opObj11.addParameter(name='xmax', value='24', format='float')
                 # opObj11.addParameter(name='zmin', value='-110', format='float')
                 # opObj11.addParameter(name='zmax', value='-70', format='float')
                 # opObj11.addParameter(name='save', value='0', format='int')
                 # # opObj11.addParameter(name='figpath', value='/tmp/', format='str')
                 #
                 opObj12 = procUnitConfObj2.addOperation(name='PublishData', optype='other')
                 opObj12.addParameter(name='zeromq', value=1, format='int')
                 # opObj13 = procUnitConfObj3.addOperation(name='PublishData', optype='other')
                 # opObj13.addParameter(name='zeromq', value=1, format='int')
                 # opObj13.addParameter(name='server', value="juanca", format='str')
-                # opObj12.addParameter(name='delay', value=1, format='int')
+                opObj12.addParameter(name='delay', value=1, format='int')
                 # print "Escribiendo el archivo XML"
                 # controllerObj.writeXml(filename)
                 # print "Leyendo el archivo XML"
                 # controllerObj.readXml(filename)
                 # timeit.timeit('controllerObj.run()', number=2)
                 controllerObj.start()
             if __name__ == '__main__':
                 parser = argparse.ArgumentParser(description='Set number of parallel processes')
-                parser.add_argument('--nProcess', default=2, type=int)
+                parser.add_argument('--nProcess', default=1, type=int)
                 args = parser.parse_args()
                 multiSchain(fiber, nProcess=args.nProcess, startDate='2015/09/26', endDate='2015/09/26')

schainpy/scripts/receiver.py 0 -1

             #!/usr/bin/env python
             '''
             Created on Jul 7, 2014
             @author: roj-idl71
             '''
             import os, sys
             from schainpy.controller import Project
             if __name__ == '__main__':
                 desc = "Segundo Test"
                 controllerObj = Project()
                 controllerObj.setup(id='191', name='test01', description=desc)
                 proc1 = controllerObj.addProcUnit(name='ReceiverData')
                 proc1.addParameter(name='realtime', value='1', format='bool')
                 proc1.addParameter(name='plottypes', value='rti', format='str')
                 proc1.addParameter(name='throttle', value='10', format='int')
                 ## TODO Agregar direccion de server de publicacion a graficos como variable
                 # op1 = proc1.addOperation(name='PlotRTIData', optype='other')
                 # op1.addParameter(name='wintitle', value='Julia 150Km', format='str')
                 # op1.addParameter(name='save', value='/home/nanosat/Pictures', format='str')
                 #
                 # op2 = proc1.addOperation(name='PlotCOHData', optype='other')
                 # op2.addParameter(name='wintitle', value='Julia 150Km', format='str')
                 # op2.addParameter(name='save', value='/home/nanosat/Pictures', format='str')
                 # #
                 # op6 = proc1.addOperation(name='PlotPHASEData', optype='other')
                 # op6.addParameter(name='wintitle', value='Julia 150Km', format='str')
                 # op6.addParameter(name='save', value='/home/nanosat/Pictures', format='str')
                 #
                 # proc2 = controllerObj.addProcUnit(name='ReceiverData')
                 # proc2.addParameter(name='server', value='juanca', format='str')
                 # proc2.addParameter(name='plottypes', value='snr,dop', format='str')
                 #
                 # op3 = proc2.addOperation(name='PlotSNRData', optype='other')
                 # op3.addParameter(name='wintitle', value='Julia 150Km', format='str')
                 # op3.addParameter(name='save', value='/home/nanosat/Pictures', format='str')
                 #
                 # op4 = proc2.addOperation(name='PlotDOPData', optype='other')
                 # op4.addParameter(name='wintitle', value='Julia 150Km', format='str')
                 # op4.addParameter(name='save', value='/home/nanosat/Pictures', format='str')
                 controllerObj.start()

schainpy/scripts/schain.xml +1 -1

	@@ -1,1 +1,1
	1	<Project description="HF_EXAMPLE" id="191" name="test01"><ReadUnit datatype="SpectraReader" id="1911" inputId="0" name="SpectraReader"><Operation id="19111" name="run" priority="1" type="self"><Parameter format="str" id="191111" name="datatype" value="SpectraReader" /><Parameter format="str" id="191112" name="path" value="/home/nanosat/data/hysell_data20/pdata" /><Parameter format="date" id="191113" name="startDate" value="2015/09/26" /><Parameter format="date" id="191114" name="endDate" value="2015/09/26" /><Parameter format="time" id="191115" name="startTime" value="00:00:00" /><Parameter format="time" id="191116" name="endTime" value="23:59:59" /><Parameter format="int" id="191118" name="cursor" value="3" /><Parameter format="int" id="191119" name="skip" value="360" /><Parameter format="int" id="191120" name="delay" value="10" /><Parameter format="int" id="191121" name="walk" value="1" /><Parameter format="int" id="191122" name="online" value="0" /></Operation></ReadUnit><ProcUnit datatype="Spectra" id="1912" inputId="1911" name="SpectraProc"><Operation id="19121" name="run" priority="1" type="self" /><Operation id="19122" name="PublishData" priority="2" type="other"><Parameter format="int" id="191221" name="zeromq" value="1" /></Operation></ProcUnit></Project> No newline at end of file		1	<Project description="HF_EXAMPLE" id="191" name="test01"><ReadUnit datatype="SpectraReader" id="1911" inputId="0" name="SpectraReader"><Operation id="19111" name="run" priority="1" type="self"><Parameter format="str" id="191111" name="datatype" value="SpectraReader" /><Parameter format="str" id="191112" name="path" value="/home/nanosat/data/hysell_data20/pdata" /><Parameter format="date" id="191113" name="startDate" value="2015/09/26" /><Parameter format="date" id="191114" name="endDate" value="2015/09/26" /><Parameter format="time" id="191115" name="startTime" value="00:00:00" /><Parameter format="time" id="191116" name="endTime" value="23:59:59" /><Parameter format="int" id="191118" name="cursor" value="2" /><Parameter format="int" id="191119" name="skip" value="720" /><Parameter format="int" id="191120" name="delay" value="10" /><Parameter format="int" id="191121" name="walk" value="1" /><Parameter format="int" id="191122" name="online" value="0" /></Operation></ReadUnit><ProcUnit datatype="Spectra" id="1912" inputId="1911" name="SpectraProc"><Operation id="19121" name="run" priority="1" type="self" /><Operation id="19122" name="PublishData" priority="2" type="other"><Parameter format="int" id="191221" name="zeromq" value="1" /><Parameter format="int" id="191222" name="delay" value="1" /></Operation></ProcUnit></Project> No newline at end of file

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