##// END OF EJS Templates
schain-jc
RSS
Change PUB/SUB to REQ/REP for send data to web server
Juan C. Espinoza -
r1161:9b21f1781b2e
parent child
Show More
Side by Side Unified
Context file:
r1161:9b21f1781b2e Loading diff...:
Show file before | Show file after | Hide comments
@@ -218,11 +218,10 class Data(object):
218
218
219 data = {}
219 data = {}
220 tm = self.times[-1]
220 tm = self.times[-1]
221
221 dy = int(self.heights.size/MAXNUMY) + 1
222 for key in self.data:
222 for key in self.data:
223 if key in ('spc', 'cspc'):
223 if key in ('spc', 'cspc'):
224 dx = int(self.data[key].shape[1]/MAXNUMX) + 1
224 dx = int(self.data[key].shape[1]/MAXNUMX) + 1
225 dy = int(self.data[key].shape[2]/MAXNUMY) + 1
226 data[key] = roundFloats(self.data[key][::, ::dx, ::dy].tolist())
225 data[key] = roundFloats(self.data[key][::, ::dx, ::dy].tolist())
227 else:
226 else:
228 data[key] = roundFloats(self.data[key][tm].tolist())
227 data[key] = roundFloats(self.data[key][tm].tolist())
@@ -232,8 +231,8 class Data(object):
232 ret['time'] = tm
231 ret['time'] = tm
233 ret['interval'] = self.interval
232 ret['interval'] = self.interval
234 ret['localtime'] = self.localtime
233 ret['localtime'] = self.localtime
235 ret['yrange'] = roundFloats(self.heights.tolist())
234 ret['yrange'] = roundFloats(self.heights[::dy].tolist())
236 if key in ('spc', 'cspc'):
235 if 'spc' in self.data or 'cspc' in self.data:
237 ret['xrange'] = roundFloats(self.xrange[2][::dx].tolist())
236 ret['xrange'] = roundFloats(self.xrange[2][::dx].tolist())
238 else:
237 else:
239 ret['xrange'] = []
238 ret['xrange'] = []
@@ -577,8 +576,10 class PlotterReceiver(ProcessingUnit, Process):
577 'Sending to web: {}'.format(self.web_address),
576 'Sending to web: {}'.format(self.web_address),
578 self.name
577 self.name
579 )
578 )
580 self.sender_web = self.context.socket(zmq.PUB)
579 self.sender_web = self.context.socket(zmq.REQ)
581 self.sender_web.connect(self.web_address)
580 self.sender_web.connect(self.web_address)
581 self.poll = zmq.Poller()
582 self.poll.register(self.sender_web, zmq.POLLIN)
582 time.sleep(1)
583 time.sleep(1)
583
584
584 if 'server' in self.kwargs:
585 if 'server' in self.kwargs:
@@ -622,12 +623,37 class PlotterReceiver(ProcessingUnit, Process):
622 if self.connections == 0 and dt in self.dates:
623 if self.connections == 0 and dt in self.dates:
623 self.data.ended = True
624 self.data.ended = True
624 self.send(self.data)
625 self.send(self.data)
625 self.data.setup()
626 # self.data.setup()
627 time.sleep(1)
628 break
626 else:
629 else:
627 if self.realtime:
630 if self.realtime:
628 self.send(self.data)
631 self.send(self.data)
629 if self.web_address:
632 if self.web_address:
630 self.sender_web.send(self.data.jsonify())
633 retries = 5
634 while True:
635 self.sender_web.send(self.data.jsonify())
636 socks = dict(self.poll.poll(5000))
637 if socks.get(self.sender_web) == zmq.POLLIN:
638 reply = self.sender_web.recv_string()
639 if reply == 'ok':
640 break
641 else:
642 print("Malformed reply from server: %s" % reply)
643
644 else:
645 print("No response from server, retrying...")
646 self.sender_web.setsockopt(zmq.LINGER, 0)
647 self.sender_web.close()
648 self.poll.unregister(self.sender_web)
649 retries -= 1
650 if retries == 0:
651 print("Server seems to be offline, abandoning")
652 break
653 self.sender_web = self.context.socket(zmq.REQ)
654 self.sender_web.connect(self.web_address)
655 self.poll.register(self.sender_web, zmq.POLLIN)
656 time.sleep(1)
631 else:
657 else:
632 self.sendData(self.send, self.data, coerce=coerce)
658 self.sendData(self.send, self.data, coerce=coerce)
633 coerce = False
659 coerce = False
General Comments 0
You need to be logged in to leave comments. Login now