schain Commit - r1235:fff8599e0346 · Jicamarca Repository

Add input queues for processing units and external operations

Juan C. Espinoza -

r1235:fff8599e0346

parent child

Context file:

r1235:fff8599e0346

Collapse all files

schainpy/model/io/bltrIO_param.py +1 -2

                     self.sizeOfFile = os.path.getsize(self.filename)
                     self.counter_records = 0
                     self.flagIsNewFile = 0
                     self.fileIndex += 1
-                    time.sleep(2)
                     return 1

schainpy/model/io/jroIO_base.py +2 -4

                     self.basicHeaderObj.size = self.basicHeaderSize  # bytes
                     self.basicHeaderObj.version = self.versionFile
                     self.basicHeaderObj.dataBlock = self.nTotalBlocks
-                    log.warning(datetime.datetime.fromtimestamp(self.dataOut.utctime))
                     utc = numpy.floor(self.dataOut.utctime)
                     milisecond = (self.dataOut.utctime - utc) * 1000.0
-                    log.warning(milisecond)
                     self.basicHeaderObj.utc = utc
                     self.basicHeaderObj.miliSecond = milisecond
                     self.basicHeaderObj.timeZone = self.dataOut.timeZone

schainpy/model/io/jroIO_param.py +2 -1

                                 data.append((dsInfo['variable'], i))
                     fp.flush()
-                    log.log('creating file: {}'.format(fp.filename), 'Writing')
+                    log.log('Creating file: {}'.format(fp.filename), self.name)
                     self.ds = dtsets
                     self.data = data
                     self.fp.flush()
                     self.blockIndex += 1
+                    log.log('Block No. {}/{}'.format(self.blockIndex, self.blocksPerFile), self.name)
                     return

schainpy/model/proc/jroproc_base.py +45 -25

             import zmq
             import time
             import pickle
+            from queue import Queue
+            from threading import Thread
             from multiprocessing import Process
             from zmq.utils.monitor import recv_monitor_message
                     return
+            class InputQueue(Thread):
+                '''
+                Class to hold input data for Proccessing Units and external Operations,
+                '''
+                def __init__(self, project_id, inputId):
+                    Thread.__init__(self)
+                    self.queue = Queue()
+                    self.project_id = project_id
+                    self.inputId = inputId
+                def run(self):
+                    c = zmq.Context()
+                    self.receiver = c.socket(zmq.SUB)
+                    self.receiver.connect(
+                        'ipc:///tmp/schain/{}_pub'.format(self.project_id))
+                    self.receiver.setsockopt(zmq.SUBSCRIBE, self.inputId.encode())
+                    while True:
+                        self.queue.put(self.receiver.recv_multipart()[1])
+                def get(self):
+                    return pickle.loads(self.queue.get())
             def MPDecorator(BaseClass):
                 """
                         self.kwargs = kwargs
                         self.sender = None
                         self.receiver = None
-                        self.i        = 0
+                        self.i = 0
                         self.name = BaseClass.__name__
                         if 'plot' in self.name.lower() and not self.name.endswith('_'):
                             self.name = '{}{}'.format(self.CODE.upper(), 'Plot')
                             self.inputId = args[0]
                             self.project_id = args[1]
                             self.typeProc = "Operation"
-                    def fix_publish(self,valor,multiple1):
+                        self.queue = InputQueue(self.project_id, self.inputId)
-                        return True if valor%multiple1 ==0 else False
                     def subscribe(self):
                         '''
-                        This function create a socket to receive objects from the
+                        This function start the input queue.
-                        topic `inputId`.
                         '''
-                        c = zmq.Context()
+                        self.queue.start()
-                        self.receiver = c.socket(zmq.SUB)
-                        self.receiver.connect(
-                            'ipc:///tmp/schain/{}_pub'.format(self.project_id))
-                        self.receiver.setsockopt(zmq.SUBSCRIBE, self.inputId.encode())
                     def listen(self):
                         '''
-                        This function waits for objects and deserialize using pickle
+                        This function waits for objects
                         '''
-                        try:
-                        	data = pickle.loads(self.receiver.recv_multipart()[1])
+                        return self.queue.get()
-                        except zmq.ZMQError as e:
-                            if e.errno == zmq.ETERM:
-                            	print (e.errno)
-                        return data
                     def set_publisher(self):
                         '''
                     def publish(self, data, id):
                         '''
                         This function publish an object, to a specific topic.
-                        The fix method only affect inputId None which is Read Unit
+                        For Read Units (inputId == None) adds a little delay
-                        Use value between 64  80, you should notice a little retard in processing
+                        to avoid data loss
                         '''
                         if self.inputId is None:
-                            self.i+=1
+                            self.i += 1
-                            if self.fix_publish(self.i,80) == True:#  value n
+                            if self.i % 100 == 0:
-                                time.sleep(0.01)
+                                self.i = 0
+                                time.sleep(0.01)
                         self.sender.send_multipart([str(id).encode(), pickle.dumps(data)])

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