##// END OF EJS Templates
schain-jc
RSS
Fix SendByFTP and add plot codes
jespinoza -
r1142:f05281601d69
parent child
Show More
Side by Side Unified
Context file:
r1142:f05281601d69 Loading diff...:
Show file before | Show file after | Hide comments
@@ -1,832 +1,835
1 '''
1 '''
2 @author: Juan C. Espinoza
2 @author: Juan C. Espinoza
3 '''
3 '''
4
4
5 import os
5 import os
6 import glob
6 import glob
7 import time
7 import time
8 import json
8 import json
9 import numpy
9 import numpy
10 import paho.mqtt.client as mqtt
10 import paho.mqtt.client as mqtt
11 import zmq
11 import zmq
12 import datetime
12 import datetime
13 import ftplib
13 import ftplib
14 from zmq.utils.monitor import recv_monitor_message
14 from zmq.utils.monitor import recv_monitor_message
15 from functools import wraps
15 from functools import wraps
16 from threading import Thread
16 from threading import Thread
17 from multiprocessing import Process
17 from multiprocessing import Process
18
18
19 from schainpy.model.proc.jroproc_base import Operation, ProcessingUnit
19 from schainpy.model.proc.jroproc_base import Operation, ProcessingUnit
20 from schainpy.model.data.jrodata import JROData
20 from schainpy.model.data.jrodata import JROData
21 from schainpy.utils import log
21 from schainpy.utils import log
22
22
23 MAXNUMX = 500
23 MAXNUMX = 500
24 MAXNUMY = 500
24 MAXNUMY = 500
25
25
26 PLOT_CODES = {
26 PLOT_CODES = {
27 'rti': 0, # Range time intensity (RTI).
27 'rti': 0, # Range time intensity (RTI).
28 'spc': 1, # Spectra (and Cross-spectra) information.
28 'spc': 1, # Spectra (and Cross-spectra) information.
29 'cspc': 2, # Cross-Correlation information.
29 'cspc': 2, # Cross-Correlation information.
30 'coh': 3, # Coherence map.
30 'coh': 3, # Coherence map.
31 'base': 4, # Base lines graphic.
31 'base': 4, # Base lines graphic.
32 'row': 5, # Row Spectra.
32 'row': 5, # Row Spectra.
33 'total' : 6, # Total Power.
33 'total': 6, # Total Power.
34 'drift' : 7, # Drifts graphics.
34 'drift': 7, # Drifts graphics.
35 'height' : 8, # Height profile.
35 'height': 8, # Height profile.
36 'phase' : 9, # Signal Phase.
36 'phase': 9, # Signal Phase.
37 'power' : 16,
37 'power': 16,
38 'noise' : 17,
38 'noise': 17,
39 'beacon' : 18,
39 'beacon': 18,
40 'wind' : 22,
40 'wind': 22,
41 'skymap' : 23,
41 'skymap': 23,
42 'V-E' : 25,
42 'Unknown': 24,
43 'Z-E' : 26,
43 'V-E': 25, # PIP Velocity.
44 'V-A' : 27,
44 'Z-E': 26, # PIP Reflectivity.
45 'Z-A' : 28,
45 'V-A': 27, # RHI Velocity.
46 'Z-A': 28, # RHI Reflectivity.
46 }
47 }
47
48
48 class PrettyFloat(float):
49 def get_plot_code(s):
49 def __repr__(self):
50 label = s.split('_')[0]
50 return '%.2f' % self
51 codes = [key for key in PLOT_CODES if key in label]
52 if codes:
53 return PLOT_CODES[codes[0]]
54 else:
55 return 24
51
56
52 def roundFloats(obj):
57 def roundFloats(obj):
53 if isinstance(obj, list):
58 if isinstance(obj, list):
54 return map(roundFloats, obj)
59 return map(roundFloats, obj)
55 elif isinstance(obj, float):
60 elif isinstance(obj, float):
56 return round(obj, 2)
61 return round(obj, 2)
57
62
58 def decimate(z, MAXNUMY):
63 def decimate(z, MAXNUMY):
59 dy = int(len(z[0])/MAXNUMY) + 1
64 dy = int(len(z[0])/MAXNUMY) + 1
60
65
61 return z[::, ::dy]
66 return z[::, ::dy]
62
67
63 class throttle(object):
68 class throttle(object):
64 '''
69 '''
65 Decorator that prevents a function from being called more than once every
70 Decorator that prevents a function from being called more than once every
66 time period.
71 time period.
67 To create a function that cannot be called more than once a minute, but
72 To create a function that cannot be called more than once a minute, but
68 will sleep until it can be called:
73 will sleep until it can be called:
69 @throttle(minutes=1)
74 @throttle(minutes=1)
70 def foo():
75 def foo():
71 pass
76 pass
72
77
73 for i in range(10):
78 for i in range(10):
74 foo()
79 foo()
75 print "This function has run %s times." % i
80 print "This function has run %s times." % i
76 '''
81 '''
77
82
78 def __init__(self, seconds=0, minutes=0, hours=0):
83 def __init__(self, seconds=0, minutes=0, hours=0):
79 self.throttle_period = datetime.timedelta(
84 self.throttle_period = datetime.timedelta(
80 seconds=seconds, minutes=minutes, hours=hours
85 seconds=seconds, minutes=minutes, hours=hours
81 )
86 )
82
87
83 self.time_of_last_call = datetime.datetime.min
88 self.time_of_last_call = datetime.datetime.min
84
89
85 def __call__(self, fn):
90 def __call__(self, fn):
86 @wraps(fn)
91 @wraps(fn)
87 def wrapper(*args, **kwargs):
92 def wrapper(*args, **kwargs):
88 coerce = kwargs.pop('coerce', None)
93 coerce = kwargs.pop('coerce', None)
89 if coerce:
94 if coerce:
90 self.time_of_last_call = datetime.datetime.now()
95 self.time_of_last_call = datetime.datetime.now()
91 return fn(*args, **kwargs)
96 return fn(*args, **kwargs)
92 else:
97 else:
93 now = datetime.datetime.now()
98 now = datetime.datetime.now()
94 time_since_last_call = now - self.time_of_last_call
99 time_since_last_call = now - self.time_of_last_call
95 time_left = self.throttle_period - time_since_last_call
100 time_left = self.throttle_period - time_since_last_call
96
101
97 if time_left > datetime.timedelta(seconds=0):
102 if time_left > datetime.timedelta(seconds=0):
98 return
103 return
99
104
100 self.time_of_last_call = datetime.datetime.now()
105 self.time_of_last_call = datetime.datetime.now()
101 return fn(*args, **kwargs)
106 return fn(*args, **kwargs)
102
107
103 return wrapper
108 return wrapper
104
109
105 class Data(object):
110 class Data(object):
106 '''
111 '''
107 Object to hold data to be plotted
112 Object to hold data to be plotted
108 '''
113 '''
109
114
110 def __init__(self, plottypes, throttle_value, exp_code, buffering=True):
115 def __init__(self, plottypes, throttle_value, exp_code, buffering=True):
111 self.plottypes = plottypes
116 self.plottypes = plottypes
112 self.throttle = throttle_value
117 self.throttle = throttle_value
113 self.exp_code = exp_code
118 self.exp_code = exp_code
114 self.buffering = buffering
119 self.buffering = buffering
115 self.ended = False
120 self.ended = False
116 self.localtime = False
121 self.localtime = False
117 self.meta = {}
122 self.meta = {}
118 self.__times = []
123 self.__times = []
119 self.__heights = []
124 self.__heights = []
120
125
121 def __str__(self):
126 def __str__(self):
122 dum = ['{}{}'.format(key, self.shape(key)) for key in self.data]
127 dum = ['{}{}'.format(key, self.shape(key)) for key in self.data]
123 return 'Data[{}][{}]'.format(';'.join(dum), len(self.__times))
128 return 'Data[{}][{}]'.format(';'.join(dum), len(self.__times))
124
129
125 def __len__(self):
130 def __len__(self):
126 return len(self.__times)
131 return len(self.__times)
127
132
128 def __getitem__(self, key):
133 def __getitem__(self, key):
129 if key not in self.data:
134 if key not in self.data:
130 raise KeyError(log.error('Missing key: {}'.format(key)))
135 raise KeyError(log.error('Missing key: {}'.format(key)))
131
136
132 if 'spc' in key or not self.buffering:
137 if 'spc' in key or not self.buffering:
133 ret = self.data[key]
138 ret = self.data[key]
134 else:
139 else:
135 ret = numpy.array([self.data[key][x] for x in self.times])
140 ret = numpy.array([self.data[key][x] for x in self.times])
136 if ret.ndim > 1:
141 if ret.ndim > 1:
137 ret = numpy.swapaxes(ret, 0, 1)
142 ret = numpy.swapaxes(ret, 0, 1)
138 return ret
143 return ret
139
144
140 def __contains__(self, key):
145 def __contains__(self, key):
141 return key in self.data
146 return key in self.data
142
147
143 def setup(self):
148 def setup(self):
144 '''
149 '''
145 Configure object
150 Configure object
146 '''
151 '''
147
152
148 self.type = ''
153 self.type = ''
149 self.ended = False
154 self.ended = False
150 self.data = {}
155 self.data = {}
151 self.__times = []
156 self.__times = []
152 self.__heights = []
157 self.__heights = []
153 self.__all_heights = set()
158 self.__all_heights = set()
154 for plot in self.plottypes:
159 for plot in self.plottypes:
155 if 'snr' in plot:
160 if 'snr' in plot:
156 plot = 'snr'
161 plot = 'snr'
157 self.data[plot] = {}
162 self.data[plot] = {}
158
163
159 def shape(self, key):
164 def shape(self, key):
160 '''
165 '''
161 Get the shape of the one-element data for the given key
166 Get the shape of the one-element data for the given key
162 '''
167 '''
163
168
164 if len(self.data[key]):
169 if len(self.data[key]):
165 if 'spc' in key or not self.buffering:
170 if 'spc' in key or not self.buffering:
166 return self.data[key].shape
171 return self.data[key].shape
167 return self.data[key][self.__times[0]].shape
172 return self.data[key][self.__times[0]].shape
168 return (0,)
173 return (0,)
169
174
170 def update(self, dataOut, tm):
175 def update(self, dataOut, tm):
171 '''
176 '''
172 Update data object with new dataOut
177 Update data object with new dataOut
173 '''
178 '''
174
179
175 if tm in self.__times:
180 if tm in self.__times:
176 return
181 return
177
182
178 self.type = dataOut.type
183 self.type = dataOut.type
179 self.parameters = getattr(dataOut, 'parameters', [])
184 self.parameters = getattr(dataOut, 'parameters', [])
180 if hasattr(dataOut, 'pairsList'):
185 if hasattr(dataOut, 'pairsList'):
181 self.pairs = dataOut.pairsList
186 self.pairs = dataOut.pairsList
182 if hasattr(dataOut, 'meta'):
187 if hasattr(dataOut, 'meta'):
183 self.meta = dataOut.meta
188 self.meta = dataOut.meta
184 self.channels = dataOut.channelList
189 self.channels = dataOut.channelList
185 self.interval = dataOut.getTimeInterval()
190 self.interval = dataOut.getTimeInterval()
186 self.localtime = dataOut.useLocalTime
191 self.localtime = dataOut.useLocalTime
187 if 'spc' in self.plottypes or 'cspc' in self.plottypes:
192 if 'spc' in self.plottypes or 'cspc' in self.plottypes:
188 self.xrange = (dataOut.getFreqRange(1)/1000., dataOut.getAcfRange(1), dataOut.getVelRange(1))
193 self.xrange = (dataOut.getFreqRange(1)/1000., dataOut.getAcfRange(1), dataOut.getVelRange(1))
189 self.__heights.append(dataOut.heightList)
194 self.__heights.append(dataOut.heightList)
190 self.__all_heights.update(dataOut.heightList)
195 self.__all_heights.update(dataOut.heightList)
191 self.__times.append(tm)
196 self.__times.append(tm)
192
197
193 for plot in self.plottypes:
198 for plot in self.plottypes:
194 if plot == 'spc':
199 if plot == 'spc':
195 z = dataOut.data_spc/dataOut.normFactor
200 z = dataOut.data_spc/dataOut.normFactor
196 self.data[plot] = 10*numpy.log10(z)
201 self.data[plot] = 10*numpy.log10(z)
197 if plot == 'cspc':
202 if plot == 'cspc':
198 self.data[plot] = dataOut.data_cspc
203 self.data[plot] = dataOut.data_cspc
199 if plot == 'noise':
204 if plot == 'noise':
200 buffer = 10*numpy.log10(dataOut.getNoise()/dataOut.normFactor)
205 buffer = 10*numpy.log10(dataOut.getNoise()/dataOut.normFactor)
201 if plot == 'rti':
206 if plot == 'rti':
202 buffer = dataOut.getPower()
207 buffer = dataOut.getPower()
203 if plot == 'snr_db':
208 if plot == 'snr_db':
204 self.data['snr'][tm] = dataOut.data_SNR
209 self.data['snr'][tm] = dataOut.data_SNR
205 if plot == 'snr':
210 if plot == 'snr':
206 buffer = 10*numpy.log10(dataOut.data_SNR)
211 buffer = 10*numpy.log10(dataOut.data_SNR)
207 if plot == 'dop':
212 if plot == 'dop':
208 buffer = 10*numpy.log10(dataOut.data_DOP)
213 buffer = 10*numpy.log10(dataOut.data_DOP)
209 if plot == 'mean':
214 if plot == 'mean':
210 buffer = dataOut.data_MEAN
215 buffer = dataOut.data_MEAN
211 if plot == 'std':
216 if plot == 'std':
212 buffer = dataOut.data_STD
217 buffer = dataOut.data_STD
213 if plot == 'coh':
218 if plot == 'coh':
214 buffer = dataOut.getCoherence()
219 buffer = dataOut.getCoherence()
215 if plot == 'phase':
220 if plot == 'phase':
216 buffer = dataOut.getCoherence(phase=True)
221 buffer = dataOut.getCoherence(phase=True)
217 if plot == 'output':
222 if plot == 'output':
218 buffer = dataOut.data_output
223 buffer = dataOut.data_output
219 if plot == 'param':
224 if plot == 'param':
220 buffer = dataOut.data_param
225 buffer = dataOut.data_param
221
226
222 if self.buffering:
227 if self.buffering:
223 self.data[plot][tm] = buffer
228 self.data[plot][tm] = buffer
224 else:
229 else:
225 self.data[plot] = buffer
230 self.data[plot] = buffer
226
231
227 def normalize_heights(self):
232 def normalize_heights(self):
228 '''
233 '''
229 Ensure same-dimension of the data for different heighList
234 Ensure same-dimension of the data for different heighList
230 '''
235 '''
231
236
232 H = numpy.array(list(self.__all_heights))
237 H = numpy.array(list(self.__all_heights))
233 H.sort()
238 H.sort()
234 for key in self.data:
239 for key in self.data:
235 shape = self.shape(key)[:-1] + H.shape
240 shape = self.shape(key)[:-1] + H.shape
236 for tm, obj in self.data[key].items():
241 for tm, obj in self.data[key].items():
237 h = self.__heights[self.__times.index(tm)]
242 h = self.__heights[self.__times.index(tm)]
238 if H.size == h.size:
243 if H.size == h.size:
239 continue
244 continue
240 index = numpy.where(numpy.in1d(H, h))[0]
245 index = numpy.where(numpy.in1d(H, h))[0]
241 dummy = numpy.zeros(shape) + numpy.nan
246 dummy = numpy.zeros(shape) + numpy.nan
242 if len(shape) == 2:
247 if len(shape) == 2:
243 dummy[:, index] = obj
248 dummy[:, index] = obj
244 else:
249 else:
245 dummy[index] = obj
250 dummy[index] = obj
246 self.data[key][tm] = dummy
251 self.data[key][tm] = dummy
247
252
248 self.__heights = [H for tm in self.__times]
253 self.__heights = [H for tm in self.__times]
249
254
250 def jsonify(self, decimate=False):
255 def jsonify(self, decimate=False):
251 '''
256 '''
252 Convert data to json
257 Convert data to json
253 '''
258 '''
254
259
255 data = {}
260 data = {}
256 tm = self.times[-1]
261 tm = self.times[-1]
257
262
258 for key in self.data:
263 for key in self.data:
259 if key in ('spc', 'cspc') or not self.buffering:
264 if key in ('spc', 'cspc') or not self.buffering:
260 dx = int(self.data[key].shape[1]/MAXNUMX) + 1
265 dx = int(self.data[key].shape[1]/MAXNUMX) + 1
261 dy = int(self.data[key].shape[2]/MAXNUMY) + 1
266 dy = int(self.data[key].shape[2]/MAXNUMY) + 1
262 data[key] = roundFloats(self.data[key][::, ::dx, ::dy].tolist())
267 data[key] = roundFloats(self.data[key][::, ::dx, ::dy].tolist())
263 else:
268 else:
264 data[key] = roundFloats(self.data[key][tm].tolist())
269 data[key] = roundFloats(self.data[key][tm].tolist())
265
270
266 ret = {'data': data}
271 ret = {'data': data}
267 ret['exp_code'] = self.exp_code
272 ret['exp_code'] = self.exp_code
268 ret['time'] = tm
273 ret['time'] = tm
269 ret['interval'] = self.interval
274 ret['interval'] = self.interval
270 ret['localtime'] = self.localtime
275 ret['localtime'] = self.localtime
271 ret['yrange'] = roundFloats(self.heights.tolist())
276 ret['yrange'] = roundFloats(self.heights.tolist())
272 if key in ('spc', 'cspc'):
277 if key in ('spc', 'cspc'):
273 ret['xrange'] = roundFloats(self.xrange[2][::dx].tolist())
278 ret['xrange'] = roundFloats(self.xrange[2][::dx].tolist())
274 else:
279 else:
275 ret['xrange'] = []
280 ret['xrange'] = []
276 if hasattr(self, 'pairs'):
281 if hasattr(self, 'pairs'):
277 ret['pairs'] = self.pairs
282 ret['pairs'] = self.pairs
278
283
279 for key, value in self.meta:
284 for key, value in self.meta:
280 ret[key] = value
285 ret[key] = value
281
286
282 return json.dumps(ret)
287 return json.dumps(ret)
283
288
284 @property
289 @property
285 def times(self):
290 def times(self):
286 '''
291 '''
287 Return the list of times of the current data
292 Return the list of times of the current data
288 '''
293 '''
289
294
290 ret = numpy.array(self.__times)
295 ret = numpy.array(self.__times)
291 ret.sort()
296 ret.sort()
292 return ret
297 return ret
293
298
294 @property
299 @property
295 def heights(self):
300 def heights(self):
296 '''
301 '''
297 Return the list of heights of the current data
302 Return the list of heights of the current data
298 '''
303 '''
299
304
300 return numpy.array(self.__heights[-1])
305 return numpy.array(self.__heights[-1])
301
306
302 class PublishData(Operation):
307 class PublishData(Operation):
303 '''
308 '''
304 Operation to send data over zmq.
309 Operation to send data over zmq.
305 '''
310 '''
306
311
307 __attrs__ = ['host', 'port', 'delay', 'zeromq', 'mqtt', 'verbose']
312 __attrs__ = ['host', 'port', 'delay', 'zeromq', 'mqtt', 'verbose']
308
313
309 def __init__(self, **kwargs):
314 def __init__(self, **kwargs):
310 """Inicio."""
315 """Inicio."""
311 Operation.__init__(self, **kwargs)
316 Operation.__init__(self, **kwargs)
312 self.isConfig = False
317 self.isConfig = False
313 self.client = None
318 self.client = None
314 self.zeromq = None
319 self.zeromq = None
315 self.mqtt = None
320 self.mqtt = None
316
321
317 def on_disconnect(self, client, userdata, rc):
322 def on_disconnect(self, client, userdata, rc):
318 if rc != 0:
323 if rc != 0:
319 log.warning('Unexpected disconnection.')
324 log.warning('Unexpected disconnection.')
320 self.connect()
325 self.connect()
321
326
322 def connect(self):
327 def connect(self):
323 log.warning('trying to connect')
328 log.warning('trying to connect')
324 try:
329 try:
325 self.client.connect(
330 self.client.connect(
326 host=self.host,
331 host=self.host,
327 port=self.port,
332 port=self.port,
328 keepalive=60*10,
333 keepalive=60*10,
329 bind_address='')
334 bind_address='')
330 self.client.loop_start()
335 self.client.loop_start()
331 # self.client.publish(
336 # self.client.publish(
332 # self.topic + 'SETUP',
337 # self.topic + 'SETUP',
333 # json.dumps(setup),
338 # json.dumps(setup),
334 # retain=True
339 # retain=True
335 # )
340 # )
336 except:
341 except:
337 log.error('MQTT Conection error.')
342 log.error('MQTT Conection error.')
338 self.client = False
343 self.client = False
339
344
340 def setup(self, port=1883, username=None, password=None, clientId="user", zeromq=1, verbose=True, **kwargs):
345 def setup(self, port=1883, username=None, password=None, clientId="user", zeromq=1, verbose=True, **kwargs):
341 self.counter = 0
346 self.counter = 0
342 self.topic = kwargs.get('topic', 'schain')
347 self.topic = kwargs.get('topic', 'schain')
343 self.delay = kwargs.get('delay', 0)
348 self.delay = kwargs.get('delay', 0)
344 self.plottype = kwargs.get('plottype', 'spectra')
349 self.plottype = kwargs.get('plottype', 'spectra')
345 self.host = kwargs.get('host', "10.10.10.82")
350 self.host = kwargs.get('host', "10.10.10.82")
346 self.port = kwargs.get('port', 3000)
351 self.port = kwargs.get('port', 3000)
347 self.clientId = clientId
352 self.clientId = clientId
348 self.cnt = 0
353 self.cnt = 0
349 self.zeromq = zeromq
354 self.zeromq = zeromq
350 self.mqtt = kwargs.get('plottype', 0)
355 self.mqtt = kwargs.get('plottype', 0)
351 self.client = None
356 self.client = None
352 self.verbose = verbose
357 self.verbose = verbose
353 setup = []
358 setup = []
354 if mqtt is 1:
359 if mqtt is 1:
355 self.client = mqtt.Client(
360 self.client = mqtt.Client(
356 client_id=self.clientId + self.topic + 'SCHAIN',
361 client_id=self.clientId + self.topic + 'SCHAIN',
357 clean_session=True)
362 clean_session=True)
358 self.client.on_disconnect = self.on_disconnect
363 self.client.on_disconnect = self.on_disconnect
359 self.connect()
364 self.connect()
360 for plot in self.plottype:
365 for plot in self.plottype:
361 setup.append({
366 setup.append({
362 'plot': plot,
367 'plot': plot,
363 'topic': self.topic + plot,
368 'topic': self.topic + plot,
364 'title': getattr(self, plot + '_' + 'title', False),
369 'title': getattr(self, plot + '_' + 'title', False),
365 'xlabel': getattr(self, plot + '_' + 'xlabel', False),
370 'xlabel': getattr(self, plot + '_' + 'xlabel', False),
366 'ylabel': getattr(self, plot + '_' + 'ylabel', False),
371 'ylabel': getattr(self, plot + '_' + 'ylabel', False),
367 'xrange': getattr(self, plot + '_' + 'xrange', False),
372 'xrange': getattr(self, plot + '_' + 'xrange', False),
368 'yrange': getattr(self, plot + '_' + 'yrange', False),
373 'yrange': getattr(self, plot + '_' + 'yrange', False),
369 'zrange': getattr(self, plot + '_' + 'zrange', False),
374 'zrange': getattr(self, plot + '_' + 'zrange', False),
370 })
375 })
371 if zeromq is 1:
376 if zeromq is 1:
372 context = zmq.Context()
377 context = zmq.Context()
373 self.zmq_socket = context.socket(zmq.PUSH)
378 self.zmq_socket = context.socket(zmq.PUSH)
374 server = kwargs.get('server', 'zmq.pipe')
379 server = kwargs.get('server', 'zmq.pipe')
375
380
376 if 'tcp://' in server:
381 if 'tcp://' in server:
377 address = server
382 address = server
378 else:
383 else:
379 address = 'ipc:///tmp/%s' % server
384 address = 'ipc:///tmp/%s' % server
380
385
381 self.zmq_socket.connect(address)
386 self.zmq_socket.connect(address)
382 time.sleep(1)
387 time.sleep(1)
383
388
384
389
385 def publish_data(self):
390 def publish_data(self):
386 self.dataOut.finished = False
391 self.dataOut.finished = False
387 if self.mqtt is 1:
392 if self.mqtt is 1:
388 yData = self.dataOut.heightList[:2].tolist()
393 yData = self.dataOut.heightList[:2].tolist()
389 if self.plottype == 'spectra':
394 if self.plottype == 'spectra':
390 data = getattr(self.dataOut, 'data_spc')
395 data = getattr(self.dataOut, 'data_spc')
391 z = data/self.dataOut.normFactor
396 z = data/self.dataOut.normFactor
392 zdB = 10*numpy.log10(z)
397 zdB = 10*numpy.log10(z)
393 xlen, ylen = zdB[0].shape
398 xlen, ylen = zdB[0].shape
394 dx = int(xlen/MAXNUMX) + 1
399 dx = int(xlen/MAXNUMX) + 1
395 dy = int(ylen/MAXNUMY) + 1
400 dy = int(ylen/MAXNUMY) + 1
396 Z = [0 for i in self.dataOut.channelList]
401 Z = [0 for i in self.dataOut.channelList]
397 for i in self.dataOut.channelList:
402 for i in self.dataOut.channelList:
398 Z[i] = zdB[i][::dx, ::dy].tolist()
403 Z[i] = zdB[i][::dx, ::dy].tolist()
399 payload = {
404 payload = {
400 'timestamp': self.dataOut.utctime,
405 'timestamp': self.dataOut.utctime,
401 'data': roundFloats(Z),
406 'data': roundFloats(Z),
402 'channels': ['Ch %s' % ch for ch in self.dataOut.channelList],
407 'channels': ['Ch %s' % ch for ch in self.dataOut.channelList],
403 'interval': self.dataOut.getTimeInterval(),
408 'interval': self.dataOut.getTimeInterval(),
404 'type': self.plottype,
409 'type': self.plottype,
405 'yData': yData
410 'yData': yData
406 }
411 }
407
412
408 elif self.plottype in ('rti', 'power'):
413 elif self.plottype in ('rti', 'power'):
409 data = getattr(self.dataOut, 'data_spc')
414 data = getattr(self.dataOut, 'data_spc')
410 z = data/self.dataOut.normFactor
415 z = data/self.dataOut.normFactor
411 avg = numpy.average(z, axis=1)
416 avg = numpy.average(z, axis=1)
412 avgdB = 10*numpy.log10(avg)
417 avgdB = 10*numpy.log10(avg)
413 xlen, ylen = z[0].shape
418 xlen, ylen = z[0].shape
414 dy = numpy.floor(ylen/self.__MAXNUMY) + 1
419 dy = numpy.floor(ylen/self.__MAXNUMY) + 1
415 AVG = [0 for i in self.dataOut.channelList]
420 AVG = [0 for i in self.dataOut.channelList]
416 for i in self.dataOut.channelList:
421 for i in self.dataOut.channelList:
417 AVG[i] = avgdB[i][::dy].tolist()
422 AVG[i] = avgdB[i][::dy].tolist()
418 payload = {
423 payload = {
419 'timestamp': self.dataOut.utctime,
424 'timestamp': self.dataOut.utctime,
420 'data': roundFloats(AVG),
425 'data': roundFloats(AVG),
421 'channels': ['Ch %s' % ch for ch in self.dataOut.channelList],
426 'channels': ['Ch %s' % ch for ch in self.dataOut.channelList],
422 'interval': self.dataOut.getTimeInterval(),
427 'interval': self.dataOut.getTimeInterval(),
423 'type': self.plottype,
428 'type': self.plottype,
424 'yData': yData
429 'yData': yData
425 }
430 }
426 elif self.plottype == 'noise':
431 elif self.plottype == 'noise':
427 noise = self.dataOut.getNoise()/self.dataOut.normFactor
432 noise = self.dataOut.getNoise()/self.dataOut.normFactor
428 noisedB = 10*numpy.log10(noise)
433 noisedB = 10*numpy.log10(noise)
429 payload = {
434 payload = {
430 'timestamp': self.dataOut.utctime,
435 'timestamp': self.dataOut.utctime,
431 'data': roundFloats(noisedB.reshape(-1, 1).tolist()),
436 'data': roundFloats(noisedB.reshape(-1, 1).tolist()),
432 'channels': ['Ch %s' % ch for ch in self.dataOut.channelList],
437 'channels': ['Ch %s' % ch for ch in self.dataOut.channelList],
433 'interval': self.dataOut.getTimeInterval(),
438 'interval': self.dataOut.getTimeInterval(),
434 'type': self.plottype,
439 'type': self.plottype,
435 'yData': yData
440 'yData': yData
436 }
441 }
437 elif self.plottype == 'snr':
442 elif self.plottype == 'snr':
438 data = getattr(self.dataOut, 'data_SNR')
443 data = getattr(self.dataOut, 'data_SNR')
439 avgdB = 10*numpy.log10(data)
444 avgdB = 10*numpy.log10(data)
440
445
441 ylen = data[0].size
446 ylen = data[0].size
442 dy = numpy.floor(ylen/self.__MAXNUMY) + 1
447 dy = numpy.floor(ylen/self.__MAXNUMY) + 1
443 AVG = [0 for i in self.dataOut.channelList]
448 AVG = [0 for i in self.dataOut.channelList]
444 for i in self.dataOut.channelList:
449 for i in self.dataOut.channelList:
445 AVG[i] = avgdB[i][::dy].tolist()
450 AVG[i] = avgdB[i][::dy].tolist()
446 payload = {
451 payload = {
447 'timestamp': self.dataOut.utctime,
452 'timestamp': self.dataOut.utctime,
448 'data': roundFloats(AVG),
453 'data': roundFloats(AVG),
449 'channels': ['Ch %s' % ch for ch in self.dataOut.channelList],
454 'channels': ['Ch %s' % ch for ch in self.dataOut.channelList],
450 'type': self.plottype,
455 'type': self.plottype,
451 'yData': yData
456 'yData': yData
452 }
457 }
453 else:
458 else:
454 print "Tipo de grafico invalido"
459 print "Tipo de grafico invalido"
455 payload = {
460 payload = {
456 'data': 'None',
461 'data': 'None',
457 'timestamp': 'None',
462 'timestamp': 'None',
458 'type': None
463 'type': None
459 }
464 }
460
465
461 self.client.publish(self.topic + self.plottype, json.dumps(payload), qos=0)
466 self.client.publish(self.topic + self.plottype, json.dumps(payload), qos=0)
462
467
463 if self.zeromq is 1:
468 if self.zeromq is 1:
464 if self.verbose:
469 if self.verbose:
465 log.log(
470 log.log(
466 'Sending {} - {}'.format(self.dataOut.type, self.dataOut.datatime),
471 'Sending {} - {}'.format(self.dataOut.type, self.dataOut.datatime),
467 self.name
472 self.name
468 )
473 )
469 self.zmq_socket.send_pyobj(self.dataOut)
474 self.zmq_socket.send_pyobj(self.dataOut)
470
475
471 def run(self, dataOut, **kwargs):
476 def run(self, dataOut, **kwargs):
472 self.dataOut = dataOut
477 self.dataOut = dataOut
473 if not self.isConfig:
478 if not self.isConfig:
474 self.setup(**kwargs)
479 self.setup(**kwargs)
475 self.isConfig = True
480 self.isConfig = True
476
481
477 self.publish_data()
482 self.publish_data()
478 time.sleep(self.delay)
483 time.sleep(self.delay)
479
484
480 def close(self):
485 def close(self):
481 if self.zeromq is 1:
486 if self.zeromq is 1:
482 self.dataOut.finished = True
487 self.dataOut.finished = True
483 self.zmq_socket.send_pyobj(self.dataOut)
488 self.zmq_socket.send_pyobj(self.dataOut)
484 time.sleep(0.1)
489 time.sleep(0.1)
485 self.zmq_socket.close()
490 self.zmq_socket.close()
486 if self.client:
491 if self.client:
487 self.client.loop_stop()
492 self.client.loop_stop()
488 self.client.disconnect()
493 self.client.disconnect()
489
494
490
495
491 class ReceiverData(ProcessingUnit):
496 class ReceiverData(ProcessingUnit):
492
497
493 __attrs__ = ['server']
498 __attrs__ = ['server']
494
499
495 def __init__(self, **kwargs):
500 def __init__(self, **kwargs):
496
501
497 ProcessingUnit.__init__(self, **kwargs)
502 ProcessingUnit.__init__(self, **kwargs)
498
503
499 self.isConfig = False
504 self.isConfig = False
500 server = kwargs.get('server', 'zmq.pipe')
505 server = kwargs.get('server', 'zmq.pipe')
501 if 'tcp://' in server:
506 if 'tcp://' in server:
502 address = server
507 address = server
503 else:
508 else:
504 address = 'ipc:///tmp/%s' % server
509 address = 'ipc:///tmp/%s' % server
505
510
506 self.address = address
511 self.address = address
507 self.dataOut = JROData()
512 self.dataOut = JROData()
508
513
509 def setup(self):
514 def setup(self):
510
515
511 self.context = zmq.Context()
516 self.context = zmq.Context()
512 self.receiver = self.context.socket(zmq.PULL)
517 self.receiver = self.context.socket(zmq.PULL)
513 self.receiver.bind(self.address)
518 self.receiver.bind(self.address)
514 time.sleep(0.5)
519 time.sleep(0.5)
515 log.success('ReceiverData from {}'.format(self.address))
520 log.success('ReceiverData from {}'.format(self.address))
516
521
517
522
518 def run(self):
523 def run(self):
519
524
520 if not self.isConfig:
525 if not self.isConfig:
521 self.setup()
526 self.setup()
522 self.isConfig = True
527 self.isConfig = True
523
528
524 self.dataOut = self.receiver.recv_pyobj()
529 self.dataOut = self.receiver.recv_pyobj()
525 log.log('{} - {}'.format(self.dataOut.type,
530 log.log('{} - {}'.format(self.dataOut.type,
526 self.dataOut.datatime.ctime(),),
531 self.dataOut.datatime.ctime(),),
527 'Receiving')
532 'Receiving')
528
533
529
534
530 class PlotterReceiver(ProcessingUnit, Process):
535 class PlotterReceiver(ProcessingUnit, Process):
531
536
532 throttle_value = 5
537 throttle_value = 5
533 __attrs__ = ['server', 'plottypes', 'realtime', 'localtime', 'throttle',
538 __attrs__ = ['server', 'plottypes', 'realtime', 'localtime', 'throttle',
534 'exp_code', 'web_server', 'buffering']
539 'exp_code', 'web_server', 'buffering']
535
540
536 def __init__(self, **kwargs):
541 def __init__(self, **kwargs):
537
542
538 ProcessingUnit.__init__(self, **kwargs)
543 ProcessingUnit.__init__(self, **kwargs)
539 Process.__init__(self)
544 Process.__init__(self)
540 self.mp = False
545 self.mp = False
541 self.isConfig = False
546 self.isConfig = False
542 self.isWebConfig = False
547 self.isWebConfig = False
543 self.connections = 0
548 self.connections = 0
544 server = kwargs.get('server', 'zmq.pipe')
549 server = kwargs.get('server', 'zmq.pipe')
545 web_server = kwargs.get('web_server', None)
550 web_server = kwargs.get('web_server', None)
546 if 'tcp://' in server:
551 if 'tcp://' in server:
547 address = server
552 address = server
548 else:
553 else:
549 address = 'ipc:///tmp/%s' % server
554 address = 'ipc:///tmp/%s' % server
550 self.address = address
555 self.address = address
551 self.web_address = web_server
556 self.web_address = web_server
552 self.plottypes = [s.strip() for s in kwargs.get('plottypes', 'rti').split(',')]
557 self.plottypes = [s.strip() for s in kwargs.get('plottypes', 'rti').split(',')]
553 self.realtime = kwargs.get('realtime', False)
558 self.realtime = kwargs.get('realtime', False)
554 self.localtime = kwargs.get('localtime', True)
559 self.localtime = kwargs.get('localtime', True)
555 self.buffering = kwargs.get('buffering', True)
560 self.buffering = kwargs.get('buffering', True)
556 self.throttle_value = kwargs.get('throttle', 5)
561 self.throttle_value = kwargs.get('throttle', 5)
557 self.exp_code = kwargs.get('exp_code', None)
562 self.exp_code = kwargs.get('exp_code', None)
558 self.sendData = self.initThrottle(self.throttle_value)
563 self.sendData = self.initThrottle(self.throttle_value)
559 self.dates = []
564 self.dates = []
560 self.setup()
565 self.setup()
561
566
562 def setup(self):
567 def setup(self):
563
568
564 self.data = Data(self.plottypes, self.throttle_value, self.exp_code, self.buffering)
569 self.data = Data(self.plottypes, self.throttle_value, self.exp_code, self.buffering)
565 self.isConfig = True
570 self.isConfig = True
566
571
567 def event_monitor(self, monitor):
572 def event_monitor(self, monitor):
568
573
569 events = {}
574 events = {}
570
575
571 for name in dir(zmq):
576 for name in dir(zmq):
572 if name.startswith('EVENT_'):
577 if name.startswith('EVENT_'):
573 value = getattr(zmq, name)
578 value = getattr(zmq, name)
574 events[value] = name
579 events[value] = name
575
580
576 while monitor.poll():
581 while monitor.poll():
577 evt = recv_monitor_message(monitor)
582 evt = recv_monitor_message(monitor)
578 if evt['event'] == 32:
583 if evt['event'] == 32:
579 self.connections += 1
584 self.connections += 1
580 if evt['event'] == 512:
585 if evt['event'] == 512:
581 pass
586 pass
582
587
583 evt.update({'description': events[evt['event']]})
588 evt.update({'description': events[evt['event']]})
584
589
585 if evt['event'] == zmq.EVENT_MONITOR_STOPPED:
590 if evt['event'] == zmq.EVENT_MONITOR_STOPPED:
586 break
591 break
587 monitor.close()
592 monitor.close()
588 print('event monitor thread done!')
593 print('event monitor thread done!')
589
594
590 def initThrottle(self, throttle_value):
595 def initThrottle(self, throttle_value):
591
596
592 @throttle(seconds=throttle_value)
597 @throttle(seconds=throttle_value)
593 def sendDataThrottled(fn_sender, data):
598 def sendDataThrottled(fn_sender, data):
594 fn_sender(data)
599 fn_sender(data)
595
600
596 return sendDataThrottled
601 return sendDataThrottled
597
602
598 def send(self, data):
603 def send(self, data):
599 log.log('Sending {}'.format(data), self.name)
604 log.log('Sending {}'.format(data), self.name)
600 self.sender.send_pyobj(data)
605 self.sender.send_pyobj(data)
601
606
602 def run(self):
607 def run(self):
603
608
604 log.log(
609 log.log(
605 'Starting from {}'.format(self.address),
610 'Starting from {}'.format(self.address),
606 self.name
611 self.name
607 )
612 )
608
613
609 self.context = zmq.Context()
614 self.context = zmq.Context()
610 self.receiver = self.context.socket(zmq.PULL)
615 self.receiver = self.context.socket(zmq.PULL)
611 self.receiver.bind(self.address)
616 self.receiver.bind(self.address)
612 monitor = self.receiver.get_monitor_socket()
617 monitor = self.receiver.get_monitor_socket()
613 self.sender = self.context.socket(zmq.PUB)
618 self.sender = self.context.socket(zmq.PUB)
614 if self.web_address:
619 if self.web_address:
615 log.success(
620 log.success(
616 'Sending to web: {}'.format(self.web_address),
621 'Sending to web: {}'.format(self.web_address),
617 self.name
622 self.name
618 )
623 )
619 self.sender_web = self.context.socket(zmq.PUSH)
624 self.sender_web = self.context.socket(zmq.PUSH)
620 self.sender_web.connect(self.web_address)
625 self.sender_web.connect(self.web_address)
621 time.sleep(1)
626 time.sleep(1)
622
627
623 if 'server' in self.kwargs:
628 if 'server' in self.kwargs:
624 self.sender.bind("ipc:///tmp/{}.plots".format(self.kwargs['server']))
629 self.sender.bind("ipc:///tmp/{}.plots".format(self.kwargs['server']))
625 else:
630 else:
626 self.sender.bind("ipc:///tmp/zmq.plots")
631 self.sender.bind("ipc:///tmp/zmq.plots")
627
632
628 time.sleep(2)
633 time.sleep(2)
629
634
630 t = Thread(target=self.event_monitor, args=(monitor,))
635 t = Thread(target=self.event_monitor, args=(monitor,))
631 t.start()
636 t.start()
632
637
633 while True:
638 while True:
634 dataOut = self.receiver.recv_pyobj()
639 dataOut = self.receiver.recv_pyobj()
635 if not dataOut.flagNoData:
640 if not dataOut.flagNoData:
636 if dataOut.type == 'Parameters':
641 if dataOut.type == 'Parameters':
637 tm = dataOut.utctimeInit
642 tm = dataOut.utctimeInit
638 else:
643 else:
639 tm = dataOut.utctime
644 tm = dataOut.utctime
640 if dataOut.useLocalTime:
645 if dataOut.useLocalTime:
641 if not self.localtime:
646 if not self.localtime:
642 tm += time.timezone
647 tm += time.timezone
643 dt = datetime.datetime.fromtimestamp(tm).date()
648 dt = datetime.datetime.fromtimestamp(tm).date()
644 else:
649 else:
645 if self.localtime:
650 if self.localtime:
646 tm -= time.timezone
651 tm -= time.timezone
647 dt = datetime.datetime.utcfromtimestamp(tm).date()
652 dt = datetime.datetime.utcfromtimestamp(tm).date()
648 coerce = False
653 coerce = False
649 if dt not in self.dates:
654 if dt not in self.dates:
650 if self.data:
655 if self.data:
651 self.data.ended = True
656 self.data.ended = True
652 self.send(self.data)
657 self.send(self.data)
653 coerce = True
658 coerce = True
654 self.data.setup()
659 self.data.setup()
655 self.dates.append(dt)
660 self.dates.append(dt)
656
661
657 self.data.update(dataOut, tm)
662 self.data.update(dataOut, tm)
658
663
659 if dataOut.finished is True:
664 if dataOut.finished is True:
660 self.connections -= 1
665 self.connections -= 1
661 if self.connections == 0 and dt in self.dates:
666 if self.connections == 0 and dt in self.dates:
662 self.data.ended = True
667 self.data.ended = True
663 self.send(self.data)
668 self.send(self.data)
664 self.data.setup()
669 self.data.setup()
665 else:
670 else:
666 if self.realtime:
671 if self.realtime:
667 self.send(self.data)
672 self.send(self.data)
668 if self.web_address:
673 if self.web_address:
669 payload = self.data.jsonify()
674 payload = self.data.jsonify()
670 log.log('Sending to web... type:{}, size:{}'.format(dataOut.type, len(payload)), self.name)
675 log.log('Sending to web... type:{}, size:{}'.format(dataOut.type, len(payload)), self.name)
671 self.sender_web.send(payload)
676 self.sender_web.send(payload)
672 else:
677 else:
673 self.sendData(self.send, self.data, coerce=coerce)
678 self.sendData(self.send, self.data, coerce=coerce)
674 coerce = False
679 coerce = False
675
680
676 return
681 return
677
682
678
683
679 class SendToFTP(Operation, Process):
684 class SendToFTP(Operation, Process):
680
685
681 '''
686 '''
682 Operation to send data over FTP.
687 Operation to send data over FTP.
683 '''
688 '''
684
689
685 __attrs__ = ['server', 'username', 'password', 'patterns', 'timeout']
690 __attrs__ = ['server', 'username', 'password', 'patterns', 'timeout']
686
691
687 def __init__(self, **kwargs):
692 def __init__(self, **kwargs):
688 '''
693 '''
689 patterns = [(local1, remote1, ext, delay, exp_code, sub_exp_code), ...]
694 patterns = [(local1, remote1, ext, delay, exp_code, sub_exp_code), ...]
690 '''
695 '''
691 Operation.__init__(self, **kwargs)
696 Operation.__init__(self, **kwargs)
692 Process.__init__(self)
697 Process.__init__(self)
693 self.server = kwargs.get('server')
698 self.server = kwargs.get('server')
694 self.username = kwargs.get('username')
699 self.username = kwargs.get('username')
695 self.password = kwargs.get('password')
700 self.password = kwargs.get('password')
696 self.patterns = kwargs.get('patterns')
701 self.patterns = kwargs.get('patterns')
697 self.timeout = kwargs.get('timeout', 30)
702 self.timeout = kwargs.get('timeout', 30)
698 self.times = [time.time() for p in self.patterns]
703 self.times = [time.time() for p in self.patterns]
699 self.latest = ['' for p in self.patterns]
704 self.latest = ['' for p in self.patterns]
700 self.mp = False
705 self.mp = False
701 self.ftp = None
706 self.ftp = None
702
707
703 def setup(self):
708 def setup(self):
704
709
705 log.log('Connecting to ftp://{}'.format(self.server), self.name)
710 log.log('Connecting to ftp://{}'.format(self.server), self.name)
706 try:
711 try:
707 self.ftp = ftplib.FTP(self.server, timeout=self.timeout)
712 self.ftp = ftplib.FTP(self.server, timeout=self.timeout)
708 except ftplib.all_errors:
713 except ftplib.all_errors:
709 log.error('Server connection fail: {}'.format(self.server), self.name)
714 log.error('Server connection fail: {}'.format(self.server), self.name)
710 if self.ftp is not None:
715 if self.ftp is not None:
711 self.ftp.close()
716 self.ftp.close()
712 self.ftp = None
717 self.ftp = None
713 self.isConfig = False
718 self.isConfig = False
714 return
719 return
715
720
716 try:
721 try:
717 self.ftp.login(self.username, self.password)
722 self.ftp.login(self.username, self.password)
718 except ftplib.all_errors:
723 except ftplib.all_errors:
719 log.error('The given username y/o password are incorrect', self.name)
724 log.error('The given username y/o password are incorrect', self.name)
720 if self.ftp is not None:
725 if self.ftp is not None:
721 self.ftp.close()
726 self.ftp.close()
722 self.ftp = None
727 self.ftp = None
723 self.isConfig = False
728 self.isConfig = False
724 return
729 return
725
730
726 log.success('Connection success', self.name)
731 log.success('Connection success', self.name)
727 self.isConfig = True
732 self.isConfig = True
728 return
733 return
729
734
730 def check(self):
735 def check(self):
731
736
732 try:
737 try:
733 self.ftp.voidcmd("NOOP")
738 self.ftp.voidcmd("NOOP")
734 except:
739 except:
735 log.warning('Connection lost... trying to reconnect', self.name)
740 log.warning('Connection lost... trying to reconnect', self.name)
736 if self.ftp is not None:
741 if self.ftp is not None:
737 self.ftp.close()
742 self.ftp.close()
738 self.ftp = None
743 self.ftp = None
739 self.setup()
744 self.setup()
740
745
741 def find_files(self, path, ext):
746 def find_files(self, path, ext):
742
747
743 files = glob.glob1(path, '*{}'.format(ext))
748 files = glob.glob1(path, '*{}'.format(ext))
744 files.sort()
749 files.sort()
745 if files:
750 if files:
746 return files[-1]
751 return files[-1]
747 return None
752 return None
748
753
749 def getftpname(self, filename, exp_code, sub_exp_code):
754 def getftpname(self, filename, exp_code, sub_exp_code):
750
755
751 thisDatetime = datetime.datetime.strptime(filename.split('_')[1], '%Y%m%d')
756 thisDatetime = datetime.datetime.strptime(filename.split('_')[1], '%Y%m%d')
752 YEAR_STR = '%4.4d'%thisDatetime.timetuple().tm_year
757 YEAR_STR = '%4.4d'%thisDatetime.timetuple().tm_year
753 DOY_STR = '%3.3d'%thisDatetime.timetuple().tm_yday
758 DOY_STR = '%3.3d'%thisDatetime.timetuple().tm_yday
754 exp_code = '%3.3d'%exp_code
759 exp_code = '%3.3d'%exp_code
755 sub_exp_code = '%2.2d'%sub_exp_code
760 sub_exp_code = '%2.2d'%sub_exp_code
756 plot_code = '%2.2d'% PLOT_CODES[filename.split('_')[0].split('-')[1]]
761 plot_code = '%2.2d'% get_plot_code(filename)
757 name = YEAR_STR + DOY_STR + '00' + exp_code + sub_exp_code + plot_code + '00.png'
762 name = YEAR_STR + DOY_STR + '00' + exp_code + sub_exp_code + plot_code + '00.png'
758 return name
763 return name
759
764
760 def upload(self, src, dst):
765 def upload(self, src, dst):
761
766
762 log.log('Uploading {} '.format(src), self.name, nl=False)
767 log.log('Uploading {} '.format(src), self.name, nl=False)
763
768
764 fp = open(src, 'rb')
769 fp = open(src, 'rb')
765 command = 'STOR {}'.format(dst)
770 command = 'STOR {}'.format(dst)
766
771
767 try:
772 try:
768 self.ftp.storbinary(command, fp, blocksize=1024)
773 self.ftp.storbinary(command, fp, blocksize=1024)
769 except ftplib.all_errors, e:
774 except Exception, e:
770 log.error('{}'.format(e), self.name)
775 log.error('{}'.format(e), self.name)
771 if self.ftp is not None:
776 if self.ftp is not None:
772 self.ftp.close()
777 self.ftp.close()
773 self.ftp = None
778 self.ftp = None
774 return
779 return 0
775
780
776 try:
781 try:
777 self.ftp.sendcmd('SITE CHMOD 755 {}'.format(dst))
782 self.ftp.sendcmd('SITE CHMOD 755 {}'.format(dst))
778 except ftplib.all_errors, e:
783 except Exception, e:
779 log.error('{}'.format(e), self.name)
784 log.error('{}'.format(e), self.name)
780 if self.ftp is not None:
785 if self.ftp is not None:
781 self.ftp.close()
786 self.ftp.close()
782 self.ftp = None
787 self.ftp = None
788 return 0
783
789
784 fp.close()
790 fp.close()
785
786 log.success('OK', tag='')
791 log.success('OK', tag='')
792 return 1
787
793
788 def send_files(self):
794 def send_files(self):
789
795
790 for x, pattern in enumerate(self.patterns):
796 for x, pattern in enumerate(self.patterns):
791 local, remote, ext, delay, exp_code, sub_exp_code = pattern
797 local, remote, ext, delay, exp_code, sub_exp_code = pattern
792 if time.time()-self.times[x] >= delay:
798 if time.time()-self.times[x] >= delay:
793 srcname = self.find_files(local, ext)
799 srcname = self.find_files(local, ext)
794
800 src = os.path.join(local, srcname)
801 if os.path.getmtime(src) < time.time() - 30*60:
802 continue
803
795 if srcname is None or srcname == self.latest[x]:
804 if srcname is None or srcname == self.latest[x]:
796 continue
805 continue
797
806
798 if 'png' in ext:
807 if 'png' in ext:
799 dstname = self.getftpname(srcname, exp_code, sub_exp_code)
808 dstname = self.getftpname(srcname, exp_code, sub_exp_code)
800 else:
809 else:
801 dstname = srcname
810 dstname = srcname
802
803 src = os.path.join(local, srcname)
804
805 if os.path.getmtime(src) < time.time() - 30*60:
806 continue
807
811
808 dst = os.path.join(remote, dstname)
812 dst = os.path.join(remote, dstname)
809
813
810 if self.ftp is None:
814 if self.upload(src, dst):
811 continue
815 self.times[x] = time.time()
812
816 self.latest[x] = srcname
813 self.upload(src, dst)
817 else:
814
818 self.isConfig = False
815 self.times[x] = time.time()
819 break
816 self.latest[x] = srcname
817
820
818 def run(self):
821 def run(self):
819
822
820 while True:
823 while True:
821 if not self.isConfig:
824 if not self.isConfig:
822 self.setup()
825 self.setup()
823 if self.ftp is not None:
826 if self.ftp is not None:
824 self.check()
827 self.check()
825 self.send_files()
828 self.send_files()
826 time.sleep(10)
829 time.sleep(10)
827
830
828 def close():
831 def close():
829
832
830 if self.ftp is not None:
833 if self.ftp is not None:
831 self.ftp.close()
834 self.ftp.close()
832 self.terminate()
835 self.terminate()
General Comments 0
You need to be logged in to leave comments. Login now