schain Commit - r1564:d1f6bedb3696 · Jicamarca Repository

Fix support for ZMQ in VoltageReader

Juan C. Espinoza -

r1564:d1f6bedb3696

parent child

Context file:

r1564:d1f6bedb3696

Collapse all files

schainpy/controller.py +3 -1

                     self.parameters = {}
                 def setup(self, project_id, id, name, datatype, err_queue, path='', startDate='', endDate='',
-                          startTime='', endTime='', server=None, **kwargs):
+                          startTime='', endTime='', server=None, topic='', **kwargs):
                     if datatype == None and name == None:
                         raise ValueError('datatype or name should be defined')
                     self.addParameter(name='endDate', value=endDate)
                     self.addParameter(name='startTime', value=startTime)
                     self.addParameter(name='endTime', value=endTime)
+                    self.addParameter(name='server', value=server)
+                    self.addParameter(name='topic', value=topic)
                     for key, value in kwargs.items():
                         self.addParameter(name=key, value=value)

schainpy/model/graphics/jroplot_base.py +1 -1

                     self.sender_time = last_time
-                    attrs = ['titles', 'zmin', 'zmax', 'tag', 'ymin', 'ymax']
+                    attrs = ['titles', 'zmin', 'zmax', 'tag', 'ymin', 'ymax', 'zlimits']
                     for attr in attrs:
                         value = getattr(self, attr)
                         if value:

schainpy/model/io/jroIO_base.py +9 -3

                 firstHeaderSize = 0
                 basicHeaderSize = 24
                 __isFirstTimeOnline = 1
+                topic = ''
                 filefmt = "*%Y%j***"
                 folderfmt = "*%Y%j"
                 __attrs__ = ['path', 'startDate', 'endDate', 'startTime', 'endTime', 'online', 'delay', 'walk']
                     if self.server is not None:
                         if 'tcp://' in self.server:
-                            address = server
+                            address = self.server
                         else:
                             address = 'ipc:///tmp/%s' % self.server
                         self.server = address
                         self.context = zmq.Context()
-                        self.receiver = self.context.socket(zmq.PULL)
+                        self.receiver = self.context.socket(zmq.SUB)
                         self.receiver.connect(self.server)
+                        self.receiver.setsockopt(zmq.SUBSCRIBE, str.encode(str(self.topic)))
                         time.sleep(0.5)
                         print('[Starting] ReceiverData from {}'.format(self.server))
                     else:
                     if self.server is None:
                         self.getData()
                     else:
-                        self.getFromServer()
+                        try:
+                            self.getFromServer()
+                        except Exception as e:
+                            log.warning('Invalid block...')
+                            self.dataOut.flagNoData = True
             class JRODataWriter(Reader):

schainpy/model/io/jroIO_voltage.py +12 -5

                     self.readFirstHeaderFromServer()
                     timestamp = self.basicHeaderObj.get_datatime()
-                    print('[Reading] - Block {} - {}'.format(self.nTotalBlocks, timestamp))
+                    print('[Receiving] - Block {} - {}'.format(self.nTotalBlocks, timestamp))
+                    if self.nTotalBlocks == self.processingHeaderObj.dataBlocksPerFile:
+                        self.nTotalBlocks = 0
+                        self.nReadBlocks = 0
+                        print('Receiving the next stream...')
                     current_pointer_location = self.blockPointer
                     junk = numpy.fromstring(
                         block[self.blockPointer:], self.dtype, self.blocksize)
                     if self.profileIndex == 0:
                         self.setBasicHeader()
-                    self.datablock[:, self.profileIndex, :] = self.dataOut.data
+                    if not self.dataOut.flagDataAsBlock:
+                        self.datablock[:, self.profileIndex, :] = self.dataOut.data
-                    self.profileIndex += 1
+                        self.profileIndex += 1
+                    else:
+                        self.datablock[:, :, :] = self.dataOut.data
+                        self.profileIndex = self.processingHeaderObj.profilesPerBlock
                     if self.hasAllDataInBuffer():
                         # if self.flagIsNewFile:
                         self.writeNextBlock()

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