##// END OF EJS Templates
schain
RSS
fix merge conflicts
Juan C. Espinoza -
r1022:0af466e13e36 merge
parent child
Show More
Side by Side Unified
Context file:
r1022:0af466e13e36 Loading diff...:
Show file before | Show file after | Hide comments
@@ -0,0 +1,113
1 import h5py
2 import numpy
3 import matplotlib.pyplot as plt
4 import glob
5 import os
6
7 #---------------------- Functions ---------------------
8
9 def findFiles(path):
10
11 dirList = []
12 fileList = []
13
14 for thisPath in os.listdir(path):
15 dirList.append(os.path.join(path,thisPath))
16 dirList.sort()
17
18 for thisDirectory in dirList:
19 files = glob.glob1(thisDirectory, "*.hdf5")
20 files.sort()
21 for thisFile in files:
22 fileList.append(os.path.join(thisDirectory,thisFile))
23
24 return fileList
25
26 def readFiles(fileList):
27
28 meteors_array = numpy.zeros((1,4))
29
30 for thisFile in fileList:
31
32 #Leer
33 f1 = h5py.File(thisFile,'r')
34 grp1 = f1['Data']
35 grp2 = grp1['data_output']
36 meteors1 = grp2['table0'][:]
37 meteors_array = numpy.vstack((meteors_array,meteors1))
38 #cerrar
39 f1.close()
40
41 meteors_array = numpy.delete(meteors_array, 0, axis=0)
42 meteors_list = [meteors_array[:,0],meteors_array[:,1],meteors_array[:,2],meteors_array[:,3]]
43 return meteors_list
44
45 def estimateMean(offset_list):
46
47 mean_off = []
48 axisY_off = []
49 axisX_off = []
50
51 for thisOffset in offset_list:
52 mean_aux = numpy.mean(thisOffset, axis = 0)
53 mean_off.append(mean_aux)
54 axisX_off.append(numpy.array([0,numpy.size(thisOffset)]))
55 axisY_off.append(numpy.array([mean_aux,mean_aux]))
56
57 return mean_off, axisY_off, axisX_off
58
59 def plotPhases(offset0, axisY0, axisX0, title):
60 f, axarr = plt.subplots(4, sharey=True)
61 color = ['b','g','r','c']
62 # plt.grid()
63 for i in range(len(offset0)):
64 thisMeteor = offset0[i]
65 thisY = axisY0[i]
66 thisX = axisX0[i]
67 thisColor = color[i]
68
69 opt = thisColor + 'o'
70 axarr[i].plot(thisMeteor,opt)
71 axarr[i].plot(thisX, thisY, thisColor)
72 axarr[i].set_ylabel('Offset ' + str(i))
73
74 plt.ylim((-180,180))
75 axarr[0].set_title(title + ' Offsets')
76 axarr[3].set_xlabel('Number of estimations')
77
78 return
79
80 def filterOffsets(offsets0, stdvLimit):
81 offsets1 = []
82
83 for thisOffset in offsets0:
84 pstd = numpy.std(thisOffset)*stdvLimit
85 pmean = numpy.mean(thisOffset)
86 outlier1 = thisOffset > pmean - pstd
87 outlier2 = thisOffset < pmean + pstd
88 not_outlier = numpy.logical_and(outlier1,outlier2)
89 thisOffset1 = thisOffset[not_outlier]
90 offsets1.append(thisOffset1)
91
92 return offsets1
93
94 #---------------------- Setup ---------------------------
95
96 path = '/home/jespinoza/Pictures/JASMET30/201608/phase'
97 stdvLimit = 0.5
98
99 #---------------------- Script ---------------------------
100
101 fileList = findFiles(path)
102 offsets0 = readFiles(fileList)
103 mean0, axisY0, axisX0 = estimateMean(offsets0)
104 plotPhases(offsets0, axisY0, axisX0, 'Original')
105
106 offsets1 = filterOffsets(offsets0, stdvLimit)
107 mean1, axisY1, axisX1 = estimateMean(offsets1)
108 plotPhases(offsets1, axisY1, axisX1, 'Filtered')
109
110 print "Original Offsets: %.2f, %.2f, %.2f, %.2f" % (mean0[0],mean0[1],mean0[2],mean0[3])
111 print "Filtered Offsets: %.2f, %.2f, %.2f, %.2f" % (mean1[0],mean1[1],mean1[2],mean1[3])
112
113 plt.show()
Show file before | Show file after | Hide comments
@@ -1,108 +1,113
1 1 # Byte-compiled / optimized / DLL files
2 2 __pycache__/
3 3 *.py[cod]
4 4 *$py.class
5 5
6 6 # C extensions
7 7 *.so
8 8
9 9 # Distribution / packaging
10 10 .Python
11 11 env/
12 12 build/
13 13 develop-eggs/
14 14 dist/
15 15 downloads/
16 16 eggs/
17 17 .eggs/
18 18 lib/
19 19 lib64/
20 20 parts/
21 21 sdist/
22 22 var/
23 23 wheels/
24 24 *.egg-info/
25 25 .installed.cfg
26 26 *.egg
27 27
28 28 # PyInstaller
29 29 # Usually these files are written by a python script from a template
30 30 # before PyInstaller builds the exe, so as to inject date/other infos into it.
31 31 *.manifest
32 32 *.spec
33 33
34 34 # Installer logs
35 35 pip-log.txt
36 36 pip-delete-this-directory.txt
37 37
38 38 # Unit test / coverage reports
39 39 htmlcov/
40 40 .tox/
41 41 .coverage
42 42 .coverage.*
43 43 .cache
44 44 nosetests.xml
45 45 coverage.xml
46 46 *,cover
47 47 .hypothesis/
48 48
49 49 # Translations
50 50 *.mo
51 51 *.pot
52 52
53 53 # Django stuff:
54 54 *.log
55 55 local_settings.py
56 56
57 57 # Flask stuff:
58 58 instance/
59 59 .webassets-cache
60 60
61 61 # Scrapy stuff:
62 62 .scrapy
63 63
64 64 # Sphinx documentation
65 65 docs/_build/
66 66
67 67 # PyBuilder
68 68 target/
69 69
70 70 # Jupyter Notebook
71 71 .ipynb_checkpoints
72 72
73 73 # pyenv
74 74 .python-version
75 75
76 76 # celery beat schedule file
77 77 celerybeat-schedule
78 78
79 79 # SageMath parsed files
80 80 *.sage.py
81 81
82 82 # dotenv
83 83 .env
84 84
85 85 # virtualenv
86 86 .venv
87 87 venv/
88 88 ENV/
89 89
90 90 # Spyder project settings
91 91 .spyderproject
92 92 .spyproject
93 93
94 94 # Rope project settings
95 95 .ropeproject
96 96
97 97 # mkdocs documentation
98 98 /site
99 99
100 100 # eclipse
101 101 .project
102 102 .pydevproject
103 103
104 104 # vscode
105 105
106 106 .vscode
107 107
108 schainpy/scripts/ No newline at end of file
108 .svn/
109 *.png
110 *.pyc
111 schainpy/scripts
112
113 schaingui/node_modules
Show file before | Show file after | Hide comments
@@ -1,132 +1,132
1 1 # Signal Chain
2 2
3 3 ## Introduction
4 4
5 5 Signal Chain (SCh) is a radar data processing library developed using [Python](www.python.org) at JRO. SCh provides modules to read, write, process and plot data.
6 6
7 7 ## Installation
8 8
9 9 Install system dependencies, clone the latest version from [git](http://jro-dev.igp.gob.pe/rhodecode/schain/) and install it as a normal python package.
10 10
11 11 ```
12 12 $ sudo apt-get install python-pip python-dev gfortran libpng-dev freetype* libblas-dev liblapack-dev libatlas-base-dev python-qt4 python-tk libssl-dev libhdf5-dev
13 13 $ sudo pip install numpy
14 14 $ git clone http://jro-dev.igp.gob.pe/rhodecode/schain/
15 15 $ cd schain
16 16 $ sudo pip install ./
17 17 ```
18 18
19 19 **Its recommended to install schain in a virtual environment**
20 20
21 21 ```
22 22 $ sudo pip install virtualenv
23 23 $ virtualenv /path/to/virtual --system-site-packages
24 24 $ source /path/to/virtual/bin/activate
25 25 (virtual) $ cd schain
26 26 (virtual) $ pip install ./
27 27 ```
28 28
29 29 ## First Script
30 30
31 31 Read Spectra data (.pdata) - remove dc - plot spectra & RTI
32 32
33 33 Import SCh and creating a project
34 34
35 35 ```python
36 36 #!/usr/bin/python
37 37
38 38 from schainpy.controller import Project
39 39
40 40 controller = Project()
41 41 controller.setup(id = '100',
42 42 name='test',
43 43 description='Basic experiment')
44 44
45 45
46 46 ```
47 47
48 48 Adding read unit and operations
49 49
50 50 ```python
51 51 read_unit = controller.addReadUnit(datatype='Spectra',
52 52 path='/path/to/pdata/',
53 53 startDate='2014/01/31',
54 54 endDate='2014/03/31',
55 55 startTime='00:00:00',
56 56 endTime='23:59:59',
57 57 online=0,
58 58 walk=0)
59 59
60 60 proc_unit = controller.addProcUnit(datatype='Spectra',
61 61 inputId=read_unit.getId())
62 62
63 63 op = proc_unit.addOperation(name='selectChannels')
64 64 op.addParameter(name='channelList', value='0,1', format='intlist')
65 65
66 66 op = proc_unit.addOperation(name='selectHeights')
67 67 op.addParameter(name='minHei', value='80', format='float')
68 68 op.addParameter(name='maxHei', value='200', format='float')
69 69
70 70 op = proc_unit.addOperation(name='removeDC')
71 71
72 72 ```
73 73
74 74 Plotting data & start project
75 75
76 76 ```python
77 77 op = proc_unit.addOperation(name='SpectraPlot', optype='other')
78 78 op.addParameter(name='id', value='1', format='int')
79 79 op.addParameter(name='wintitle', value='Spectra', format='str')
80 80
81 81 op = procUnitConfObj1.addOperation(name='RTIPlot', optype='other')
82 82 op.addParameter(name='id', value='2', format='int')
83 83 op.addParameter(name='wintitle', value='RTI', format='str')
84 84
85 85 controller.start()
86 86
87 87 ```
88 88
89 89 Full script
90 90
91 91
92 92 ```python
93 93 #!/usr/bin/python
94 94
95 95 from schainpy.controller import Project
96 96
97 97 controller = Project()
98 98 controller.setup(id = '100',
99 99 name='test',
100 100 description='Basic experiment')
101 101 read_unit = controller.addReadUnit(datatype='Spectra',
102 102 path='/path/to/pdata/',
103 103 startDate='2014/01/31',
104 104 endDate='2014/03/31',
105 105 startTime='00:00:00',
106 106 endTime='23:59:59',
107 107 online=0,
108 108 walk=0)
109 109
110 110 proc_unit = controller.addProcUnit(datatype='Spectra',
111 111 inputId=read_unit.getId())
112 112
113 113 op = proc_unit.addOperation(name='selectChannels')
114 114 op.addParameter(name='channelList', value='0,1', format='intlist')
115 115
116 116 op = proc_unit.addOperation(name='selectHeights')
117 117 op.addParameter(name='minHei', value='80', format='float')
118 118 op.addParameter(name='maxHei', value='200', format='float')
119 119
120 120 op = proc_unit.addOperation(name='removeDC')
121 121
122 122 op = proc_unit.addOperation(name='SpectraPlot', optype='other')
123 123 op.addParameter(name='id', value='6', format='int')
124 124 op.addParameter(name='wintitle', value='Spectra', format='str')
125 125
126 126 op = procUnitConfObj1.addOperation(name='RTIPlot', optype='other')
127 127 op.addParameter(name='id', value='2', format='int')
128 128 op.addParameter(name='wintitle', value='RTI', format='str')
129 129
130 130 controller.start()
131 131
132 ```
132 ``` No newline at end of file
Show file before | Show file after | Hide comments
@@ -1,1328 +1,1328
1 1 '''
2 2 Created on September , 2012
3 3 @author:
4 4 '''
5 5
6 6 import sys
7 7 import ast
8 8 import datetime
9 9 import traceback
10 10 import math
11 11 import time
12 12 from multiprocessing import Process, Queue, cpu_count
13 13
14 14 import schainpy
15 15 import schainpy.admin
16 16
17 17 from xml.etree.ElementTree import ElementTree, Element, SubElement, tostring
18 18 from xml.dom import minidom
19 19
20 20 from schainpy.model import *
21 21 from time import sleep
22 22
23 23 def prettify(elem):
24 24 """Return a pretty-printed XML string for the Element.
25 25 """
26 26 rough_string = tostring(elem, 'utf-8')
27 27 reparsed = minidom.parseString(rough_string)
28 28 return reparsed.toprettyxml(indent=" ")
29 29
30 30 def multiSchain(child, nProcess=cpu_count(), startDate=None, endDate=None, by_day=False):
31 31 skip = 0
32 32 cursor = 0
33 33 nFiles = None
34 34 processes = []
35 35 dt1 = datetime.datetime.strptime(startDate, '%Y/%m/%d')
36 36 dt2 = datetime.datetime.strptime(endDate, '%Y/%m/%d')
37 37 days = (dt2 - dt1).days
38 38
39 39 for day in range(days+1):
40 40 skip = 0
41 41 cursor = 0
42 42 q = Queue()
43 43 processes = []
44 44 dt = (dt1 + datetime.timedelta(day)).strftime('%Y/%m/%d')
45 45 firstProcess = Process(target=child, args=(cursor, skip, q, dt))
46 46 firstProcess.start()
47 47 if by_day:
48 48 continue
49 49 nFiles = q.get()
50 50 if nFiles==0:
51 51 continue
52 52 firstProcess.terminate()
53 53 skip = int(math.ceil(nFiles/nProcess))
54 54 while True:
55 55 processes.append(Process(target=child, args=(cursor, skip, q, dt)))
56 56 processes[cursor].start()
57 57 if nFiles < cursor*skip:
58 58 break
59 59 cursor += 1
60 60
61 61 def beforeExit(exctype, value, trace):
62 62 for process in processes:
63 63 process.terminate()
64 64 process.join()
65 65 print traceback.print_tb(trace)
66 66
67 67 sys.excepthook = beforeExit
68 68
69 69 for process in processes:
70 70 process.join()
71 71 process.terminate()
72 72
73 73 time.sleep(3)
74 74
75 75
76 76 class ParameterConf():
77 77
78 78 id = None
79 79 name = None
80 80 value = None
81 81 format = None
82 82
83 83 __formated_value = None
84 84
85 85 ELEMENTNAME = 'Parameter'
86 86
87 87 def __init__(self):
88 88
89 89 self.format = 'str'
90 90
91 91 def getElementName(self):
92 92
93 93 return self.ELEMENTNAME
94 94
95 95 def getValue(self):
96 96
97 97 value = self.value
98 98 format = self.format
99 99
100 100 if self.__formated_value != None:
101 101
102 102 return self.__formated_value
103 103
104 104 if format == 'obj':
105 105 return value
106 106
107 107 if format == 'str':
108 108 self.__formated_value = str(value)
109 109 return self.__formated_value
110 110
111 111 if value == '':
112 112 raise ValueError, "%s: This parameter value is empty" %self.name
113 113
114 114 if format == 'list':
115 115 strList = value.split(',')
116 116
117 117 self.__formated_value = strList
118 118
119 119 return self.__formated_value
120 120
121 121 if format == 'intlist':
122 122 """
123 123 Example:
124 124 value = (0,1,2)
125 125 """
126 126
127 127 new_value = ast.literal_eval(value)
128 128
129 129 if type(new_value) not in (tuple, list):
130 130 new_value = [int(new_value)]
131 131
132 132 self.__formated_value = new_value
133 133
134 134 return self.__formated_value
135 135
136 136 if format == 'floatlist':
137 137 """
138 138 Example:
139 139 value = (0.5, 1.4, 2.7)
140 140 """
141 141
142 142 new_value = ast.literal_eval(value)
143 143
144 144 if type(new_value) not in (tuple, list):
145 145 new_value = [float(new_value)]
146 146
147 147 self.__formated_value = new_value
148 148
149 149 return self.__formated_value
150 150
151 151 if format == 'date':
152 152 strList = value.split('/')
153 153 intList = [int(x) for x in strList]
154 154 date = datetime.date(intList[0], intList[1], intList[2])
155 155
156 156 self.__formated_value = date
157 157
158 158 return self.__formated_value
159 159
160 160 if format == 'time':
161 161 strList = value.split(':')
162 162 intList = [int(x) for x in strList]
163 163 time = datetime.time(intList[0], intList[1], intList[2])
164 164
165 165 self.__formated_value = time
166 166
167 167 return self.__formated_value
168 168
169 169 if format == 'pairslist':
170 170 """
171 171 Example:
172 172 value = (0,1),(1,2)
173 173 """
174 174
175 175 new_value = ast.literal_eval(value)
176 176
177 177 if type(new_value) not in (tuple, list):
178 178 raise ValueError, "%s has to be a tuple or list of pairs" %value
179 179
180 180 if type(new_value[0]) not in (tuple, list):
181 181 if len(new_value) != 2:
182 182 raise ValueError, "%s has to be a tuple or list of pairs" %value
183 183 new_value = [new_value]
184 184
185 185 for thisPair in new_value:
186 186 if len(thisPair) != 2:
187 187 raise ValueError, "%s has to be a tuple or list of pairs" %value
188 188
189 189 self.__formated_value = new_value
190 190
191 191 return self.__formated_value
192 192
193 193 if format == 'multilist':
194 194 """
195 195 Example:
196 196 value = (0,1,2),(3,4,5)
197 197 """
198 198 multiList = ast.literal_eval(value)
199 199
200 200 if type(multiList[0]) == int:
201 201 multiList = ast.literal_eval("(" + value + ")")
202 202
203 203 self.__formated_value = multiList
204 204
205 205 return self.__formated_value
206 206
207 207 if format == 'bool':
208 208 value = int(value)
209 209
210 210 if format == 'int':
211 211 value = float(value)
212 212
213 213 format_func = eval(format)
214 214
215 215 self.__formated_value = format_func(value)
216 216
217 217 return self.__formated_value
218 218
219 219 def updateId(self, new_id):
220 220
221 221 self.id = str(new_id)
222 222
223 223 def setup(self, id, name, value, format='str'):
224 224 self.id = str(id)
225 225 self.name = name
226 226 if format == 'obj':
227 227 self.value = value
228 228 else:
229 229 self.value = str(value)
230 230 self.format = str.lower(format)
231 231
232 232 self.getValue()
233 233
234 234 return 1
235 235
236 236 def update(self, name, value, format='str'):
237 237
238 238 self.name = name
239 239 self.value = str(value)
240 240 self.format = format
241 241
242 242 def makeXml(self, opElement):
243 243 if self.name not in ('queue',):
244 244 parmElement = SubElement(opElement, self.ELEMENTNAME)
245 245 parmElement.set('id', str(self.id))
246 246 parmElement.set('name', self.name)
247 247 parmElement.set('value', self.value)
248 248 parmElement.set('format', self.format)
249 249
250 250 def readXml(self, parmElement):
251 251
252 252 self.id = parmElement.get('id')
253 253 self.name = parmElement.get('name')
254 254 self.value = parmElement.get('value')
255 255 self.format = str.lower(parmElement.get('format'))
256 256
257 257 #Compatible with old signal chain version
258 258 if self.format == 'int' and self.name == 'idfigure':
259 259 self.name = 'id'
260 260
261 261 def printattr(self):
262 262
263 263 print "Parameter[%s]: name = %s, value = %s, format = %s" %(self.id, self.name, self.value, self.format)
264 264
265 265 class OperationConf():
266 266
267 267 id = None
268 268 name = None
269 269 priority = None
270 270 type = None
271 271
272 272 parmConfObjList = []
273 273
274 274 ELEMENTNAME = 'Operation'
275 275
276 276 def __init__(self):
277 277
278 278 self.id = '0'
279 279 self.name = None
280 280 self.priority = None
281 281 self.type = 'self'
282 282
283 283
284 284 def __getNewId(self):
285 285
286 286 return int(self.id)*10 + len(self.parmConfObjList) + 1
287 287
288 288 def updateId(self, new_id):
289 289
290 290 self.id = str(new_id)
291 291
292 292 n = 1
293 293 for parmObj in self.parmConfObjList:
294 294
295 295 idParm = str(int(new_id)*10 + n)
296 296 parmObj.updateId(idParm)
297 297
298 298 n += 1
299 299
300 300 def getElementName(self):
301 301
302 302 return self.ELEMENTNAME
303 303
304 304 def getParameterObjList(self):
305 305
306 306 return self.parmConfObjList
307 307
308 308 def getParameterObj(self, parameterName):
309 309
310 310 for parmConfObj in self.parmConfObjList:
311 311
312 312 if parmConfObj.name != parameterName:
313 313 continue
314 314
315 315 return parmConfObj
316 316
317 317 return None
318 318
319 319 def getParameterObjfromValue(self, parameterValue):
320 320
321 321 for parmConfObj in self.parmConfObjList:
322 322
323 323 if parmConfObj.getValue() != parameterValue:
324 324 continue
325 325
326 326 return parmConfObj.getValue()
327 327
328 328 return None
329 329
330 330 def getParameterValue(self, parameterName):
331 331
332 332 parameterObj = self.getParameterObj(parameterName)
333
334 # if not parameterObj:
335 # return None
336
333
334 # if not parameterObj:
335 # return None
336
337 337 value = parameterObj.getValue()
338 338
339 339 return value
340 340
341 341
342 342 def getKwargs(self):
343 343
344 344 kwargs = {}
345 345
346 346 for parmConfObj in self.parmConfObjList:
347 347 if self.name == 'run' and parmConfObj.name == 'datatype':
348 348 continue
349 349
350 350 kwargs[parmConfObj.name] = parmConfObj.getValue()
351 351
352 352 return kwargs
353 353
354 354 def setup(self, id, name, priority, type):
355 355
356 356 self.id = str(id)
357 357 self.name = name
358 358 self.type = type
359 359 self.priority = priority
360 360
361 361 self.parmConfObjList = []
362 362
363 363 def removeParameters(self):
364 364
365 365 for obj in self.parmConfObjList:
366 366 del obj
367 367
368 368 self.parmConfObjList = []
369 369
370 370 def addParameter(self, name, value, format='str'):
371 371
372 372 id = self.__getNewId()
373 373
374 374 parmConfObj = ParameterConf()
375 375 if not parmConfObj.setup(id, name, value, format):
376 376 return None
377 377
378 378 self.parmConfObjList.append(parmConfObj)
379 379
380 380 return parmConfObj
381 381
382 382 def changeParameter(self, name, value, format='str'):
383 383
384 384 parmConfObj = self.getParameterObj(name)
385 385 parmConfObj.update(name, value, format)
386 386
387 387 return parmConfObj
388 388
389 389 def makeXml(self, procUnitElement):
390 390
391 391 opElement = SubElement(procUnitElement, self.ELEMENTNAME)
392 392 opElement.set('id', str(self.id))
393 393 opElement.set('name', self.name)
394 394 opElement.set('type', self.type)
395 395 opElement.set('priority', str(self.priority))
396 396
397 397 for parmConfObj in self.parmConfObjList:
398 398 parmConfObj.makeXml(opElement)
399 399
400 400 def readXml(self, opElement):
401 401
402 402 self.id = opElement.get('id')
403 403 self.name = opElement.get('name')
404 404 self.type = opElement.get('type')
405 405 self.priority = opElement.get('priority')
406 406
407 407 #Compatible with old signal chain version
408 408 #Use of 'run' method instead 'init'
409 409 if self.type == 'self' and self.name == 'init':
410 410 self.name = 'run'
411 411
412 412 self.parmConfObjList = []
413 413
414 414 parmElementList = opElement.iter(ParameterConf().getElementName())
415 415
416 416 for parmElement in parmElementList:
417 417 parmConfObj = ParameterConf()
418 418 parmConfObj.readXml(parmElement)
419 419
420 420 #Compatible with old signal chain version
421 421 #If an 'plot' OPERATION is found, changes name operation by the value of its type PARAMETER
422 422 if self.type != 'self' and self.name == 'Plot':
423 423 if parmConfObj.format == 'str' and parmConfObj.name == 'type':
424 424 self.name = parmConfObj.value
425 425 continue
426 426
427 427 self.parmConfObjList.append(parmConfObj)
428 428
429 429 def printattr(self):
430 430
431 431 print "%s[%s]: name = %s, type = %s, priority = %s" %(self.ELEMENTNAME,
432 432 self.id,
433 433 self.name,
434 434 self.type,
435 435 self.priority)
436 436
437 437 for parmConfObj in self.parmConfObjList:
438 438 parmConfObj.printattr()
439 439
440 440 def createObject(self, plotter_queue=None):
441 441
442 442
443 443 if self.type == 'self':
444 444 raise ValueError, "This operation type cannot be created"
445 445
446 446 if self.type == 'plotter':
447 447 #Plotter(plotter_name)
448 448 if not plotter_queue:
449 449 raise ValueError, "plotter_queue is not defined. Use:\nmyProject = Project()\nmyProject.setPlotterQueue(plotter_queue)"
450 450
451 451 opObj = Plotter(self.name, plotter_queue)
452 452
453 453 if self.type == 'external' or self.type == 'other':
454 454
455 455 className = eval(self.name)
456 456 kwargs = self.getKwargs()
457 457
458 458 opObj = className(**kwargs)
459 459
460 460 return opObj
461 461
462 462
463 463 class ProcUnitConf():
464 464
465 465 id = None
466 466 name = None
467 467 datatype = None
468 468 inputId = None
469 469 parentId = None
470 470
471 471 opConfObjList = []
472 472
473 473 procUnitObj = None
474 474 opObjList = []
475 475
476 476 ELEMENTNAME = 'ProcUnit'
477 477
478 478 def __init__(self):
479 479
480 480 self.id = None
481 481 self.datatype = None
482 482 self.name = None
483 483 self.inputId = None
484 484
485 485 self.opConfObjList = []
486 486
487 487 self.procUnitObj = None
488 488 self.opObjDict = {}
489 489
490 490 def __getPriority(self):
491 491
492 492 return len(self.opConfObjList)+1
493 493
494 494 def __getNewId(self):
495 495
496 496 return int(self.id)*10 + len(self.opConfObjList) + 1
497 497
498 498 def getElementName(self):
499 499
500 500 return self.ELEMENTNAME
501 501
502 502 def getId(self):
503 503
504 504 return self.id
505 505
506 506 def updateId(self, new_id, parentId=parentId):
507 507
508 508
509 509 new_id = int(parentId)*10 + (int(self.id) % 10)
510 510 new_inputId = int(parentId)*10 + (int(self.inputId) % 10)
511 511
512 512 #If this proc unit has not inputs
513 513 if self.inputId == '0':
514 514 new_inputId = 0
515 515
516 516 n = 1
517 517 for opConfObj in self.opConfObjList:
518 518
519 519 idOp = str(int(new_id)*10 + n)
520 520 opConfObj.updateId(idOp)
521 521
522 522 n += 1
523 523
524 524 self.parentId = str(parentId)
525 525 self.id = str(new_id)
526 526 self.inputId = str(new_inputId)
527 527
528 528
529 529 def getInputId(self):
530 530
531 531 return self.inputId
532 532
533 533 def getOperationObjList(self):
534 534
535 535 return self.opConfObjList
536 536
537 537 def getOperationObj(self, name=None):
538 538
539 539 for opConfObj in self.opConfObjList:
540 540
541 541 if opConfObj.name != name:
542 542 continue
543 543
544 544 return opConfObj
545 545
546 546 return None
547 547
548 548 def getOpObjfromParamValue(self, value=None):
549 549
550 550 for opConfObj in self.opConfObjList:
551 551 if opConfObj.getParameterObjfromValue(parameterValue=value) != value:
552 552 continue
553 553 return opConfObj
554 554 return None
555 555
556 556 def getProcUnitObj(self):
557 557
558 558 return self.procUnitObj
559 559
560 560 def setup(self, id, name, datatype, inputId, parentId=None):
561 561
562 562 #Compatible with old signal chain version
563 563 if datatype==None and name==None:
564 564 raise ValueError, "datatype or name should be defined"
565 565
566 566 if name==None:
567 567 if 'Proc' in datatype:
568 568 name = datatype
569 569 else:
570 570 name = '%sProc' %(datatype)
571 571
572 572 if datatype==None:
573 573 datatype = name.replace('Proc','')
574 574
575 575 self.id = str(id)
576 576 self.name = name
577 577 self.datatype = datatype
578 578 self.inputId = inputId
579 579 self.parentId = parentId
580 580
581 581 self.opConfObjList = []
582 582
583 583 self.addOperation(name='run', optype='self')
584 584
585 585 def removeOperations(self):
586 586
587 587 for obj in self.opConfObjList:
588 588 del obj
589 589
590 590 self.opConfObjList = []
591 591 self.addOperation(name='run')
592 592
593 593 def addParameter(self, **kwargs):
594 594 '''
595 595 Add parameters to "run" operation
596 596 '''
597 597 opObj = self.opConfObjList[0]
598 598
599 599 opObj.addParameter(**kwargs)
600 600
601 601 return opObj
602 602
603 603 def addOperation(self, name, optype='self'):
604 604
605 605 id = self.__getNewId()
606 606 priority = self.__getPriority()
607 607
608 608 opConfObj = OperationConf()
609 609 opConfObj.setup(id, name=name, priority=priority, type=optype)
610 610
611 611 self.opConfObjList.append(opConfObj)
612 612
613 613 return opConfObj
614 614
615 615 def makeXml(self, projectElement):
616 616
617 617 procUnitElement = SubElement(projectElement, self.ELEMENTNAME)
618 618 procUnitElement.set('id', str(self.id))
619 619 procUnitElement.set('name', self.name)
620 620 procUnitElement.set('datatype', self.datatype)
621 621 procUnitElement.set('inputId', str(self.inputId))
622 622
623 623 for opConfObj in self.opConfObjList:
624 624 opConfObj.makeXml(procUnitElement)
625 625
626 626 def readXml(self, upElement):
627 627
628 628 self.id = upElement.get('id')
629 629 self.name = upElement.get('name')
630 630 self.datatype = upElement.get('datatype')
631 631 self.inputId = upElement.get('inputId')
632 632
633 633 if self.ELEMENTNAME == "ReadUnit":
634 634 self.datatype = self.datatype.replace("Reader", "")
635 635
636 636 if self.ELEMENTNAME == "ProcUnit":
637 637 self.datatype = self.datatype.replace("Proc", "")
638 638
639 639 if self.inputId == 'None':
640 640 self.inputId = '0'
641 641
642 642 self.opConfObjList = []
643 643
644 644 opElementList = upElement.iter(OperationConf().getElementName())
645 645
646 646 for opElement in opElementList:
647 647 opConfObj = OperationConf()
648 648 opConfObj.readXml(opElement)
649 649 self.opConfObjList.append(opConfObj)
650 650
651 651 def printattr(self):
652 652
653 653 print "%s[%s]: name = %s, datatype = %s, inputId = %s" %(self.ELEMENTNAME,
654 self.id,
655 self.name,
656 self.datatype,
657 self.inputId)
658
654 self.id,
655 self.name,
656 self.datatype,
657 self.inputId)
658
659 659 for opConfObj in self.opConfObjList:
660 660 opConfObj.printattr()
661 661
662 662
663 663 def getKwargs(self):
664 664
665 665 opObj = self.opConfObjList[0]
666 666 kwargs = opObj.getKwargs()
667 667
668 668 return kwargs
669 669
670 670 def createObjects(self, plotter_queue=None):
671 671
672 672 className = eval(self.name)
673 673 kwargs = self.getKwargs()
674 674 procUnitObj = className(**kwargs)
675 675
676 676 for opConfObj in self.opConfObjList:
677 677
678 678 if opConfObj.type=='self' and self.name=='run':
679 679 continue
680 680 elif opConfObj.type=='self':
681 681 procUnitObj.addOperationKwargs(opConfObj.id, **opConfObj.getKwargs())
682 682 continue
683 683
684 684 opObj = opConfObj.createObject(plotter_queue)
685 685
686 686 self.opObjDict[opConfObj.id] = opObj
687 687
688 688 procUnitObj.addOperation(opObj, opConfObj.id)
689 689
690 690 self.procUnitObj = procUnitObj
691 691
692 692 return procUnitObj
693 693
694 694 def run(self):
695 695
696 696 is_ok = False
697 697
698 698 for opConfObj in self.opConfObjList:
699 699
700 700 kwargs = {}
701 701 for parmConfObj in opConfObj.getParameterObjList():
702 702 if opConfObj.name == 'run' and parmConfObj.name == 'datatype':
703 703 continue
704 704
705 705 kwargs[parmConfObj.name] = parmConfObj.getValue()
706 706
707 707 #ini = time.time()
708 708
709 709 #print "\tRunning the '%s' operation with %s" %(opConfObj.name, opConfObj.id)
710 710 sts = self.procUnitObj.call(opType = opConfObj.type,
711 711 opName = opConfObj.name,
712 712 opId = opConfObj.id,
713 )
714
715 # total_time = time.time() - ini
716 #
717 # if total_time > 0.002:
718 # print "%s::%s took %f seconds" %(self.name, opConfObj.name, total_time)
719
713 **kwargs)
714
715 # total_time = time.time() - ini
716 #
717 # if total_time > 0.002:
718 # print "%s::%s took %f seconds" %(self.name, opConfObj.name, total_time)
719
720 720 is_ok = is_ok or sts
721 721
722 722 return is_ok
723 723
724 724 def close(self):
725 725
726 726 for opConfObj in self.opConfObjList:
727 727 if opConfObj.type == 'self':
728 728 continue
729 729
730 730 opObj = self.procUnitObj.getOperationObj(opConfObj.id)
731 731 opObj.close()
732 732
733 733 self.procUnitObj.close()
734 734
735 735 return
736 736
737 737 class ReadUnitConf(ProcUnitConf):
738 738
739 739 path = None
740 740 startDate = None
741 741 endDate = None
742 742 startTime = None
743 743 endTime = None
744 744
745 745 ELEMENTNAME = 'ReadUnit'
746 746
747 747 def __init__(self):
748 748
749 749 self.id = None
750 750 self.datatype = None
751 751 self.name = None
752 752 self.inputId = None
753 753
754 754 self.parentId = None
755 755
756 756 self.opConfObjList = []
757 757 self.opObjList = []
758 758
759 759 def getElementName(self):
760 760
761 761 return self.ELEMENTNAME
762 762
763 763 def setup(self, id, name, datatype, path='', startDate="", endDate="", startTime="",
764 764 endTime="", parentId=None, queue=None, server=None, **kwargs):
765 765
766 766 #Compatible with old signal chain version
767 767 if datatype==None and name==None:
768 768 raise ValueError, "datatype or name should be defined"
769 769
770 770 if name==None:
771 771 if 'Reader' in datatype:
772 772 name = datatype
773 773 else:
774 774 name = '%sReader' %(datatype)
775 775 if datatype==None:
776 776 datatype = name.replace('Reader','')
777 777
778 778 self.id = id
779 779 self.name = name
780 780 self.datatype = datatype
781 781 if path != '':
782 782 self.path = os.path.abspath(path)
783 783 self.startDate = startDate
784 784 self.endDate = endDate
785 785 self.startTime = startTime
786 786 self.endTime = endTime
787 787
788 788 self.inputId = '0'
789 789 self.parentId = parentId
790 790 self.queue = queue
791 791 self.server = server
792 792 self.addRunOperation(**kwargs)
793 793
794 794 def update(self, datatype, path, startDate, endDate, startTime, endTime, parentId=None, name=None, **kwargs):
795 795
796 796 #Compatible with old signal chain version
797 797 if datatype==None and name==None:
798 798 raise ValueError, "datatype or name should be defined"
799 799
800 800 if name==None:
801 801 if 'Reader' in datatype:
802 802 name = datatype
803 803 else:
804 804 name = '%sReader' %(datatype)
805 805
806 806 if datatype==None:
807 807 datatype = name.replace('Reader','')
808 808
809 809 self.datatype = datatype
810 810 self.name = name
811 811 self.path = path
812 812 self.startDate = startDate
813 813 self.endDate = endDate
814 814 self.startTime = startTime
815 815 self.endTime = endTime
816 816
817 817 self.inputId = '0'
818 818 self.parentId = parentId
819 819
820 820 self.updateRunOperation(**kwargs)
821 821
822 822 def removeOperations(self):
823 823
824 824 for obj in self.opConfObjList:
825 825 del obj
826 826
827 827 self.opConfObjList = []
828 828
829 829 def addRunOperation(self, **kwargs):
830 830
831 831 opObj = self.addOperation(name = 'run', optype = 'self')
832 832
833 833 if self.server is None:
834 834 opObj.addParameter(name='datatype' , value=self.datatype, format='str')
835 835 opObj.addParameter(name='path' , value=self.path, format='str')
836 836 opObj.addParameter(name='startDate' , value=self.startDate, format='date')
837 837 opObj.addParameter(name='endDate' , value=self.endDate, format='date')
838 838 opObj.addParameter(name='startTime' , value=self.startTime, format='time')
839 839 opObj.addParameter(name='endTime' , value=self.endTime, format='time')
840 840 opObj.addParameter(name='queue' , value=self.queue, format='obj')
841 841 for key, value in kwargs.items():
842 842 opObj.addParameter(name=key, value=value, format=type(value).__name__)
843 843 else:
844 844 opObj.addParameter(name='server' , value=self.server, format='str')
845 845
846 846
847 847 return opObj
848 848
849 849 def updateRunOperation(self, **kwargs):
850 850
851 851 opObj = self.getOperationObj(name = 'run')
852 852 opObj.removeParameters()
853 853
854 854 opObj.addParameter(name='datatype' , value=self.datatype, format='str')
855 855 opObj.addParameter(name='path' , value=self.path, format='str')
856 856 opObj.addParameter(name='startDate' , value=self.startDate, format='date')
857 857 opObj.addParameter(name='endDate' , value=self.endDate, format='date')
858 858 opObj.addParameter(name='startTime' , value=self.startTime, format='time')
859 859 opObj.addParameter(name='endTime' , value=self.endTime, format='time')
860 860
861 861 for key, value in kwargs.items():
862 862 opObj.addParameter(name=key, value=value, format=type(value).__name__)
863 863
864 864 return opObj
865
866 # def makeXml(self, projectElement):
867 #
868 # procUnitElement = SubElement(projectElement, self.ELEMENTNAME)
869 # procUnitElement.set('id', str(self.id))
870 # procUnitElement.set('name', self.name)
871 # procUnitElement.set('datatype', self.datatype)
872 # procUnitElement.set('inputId', str(self.inputId))
873 #
874 # for opConfObj in self.opConfObjList:
875 # opConfObj.makeXml(procUnitElement)
876
865
866 # def makeXml(self, projectElement):
867 #
868 # procUnitElement = SubElement(projectElement, self.ELEMENTNAME)
869 # procUnitElement.set('id', str(self.id))
870 # procUnitElement.set('name', self.name)
871 # procUnitElement.set('datatype', self.datatype)
872 # procUnitElement.set('inputId', str(self.inputId))
873 #
874 # for opConfObj in self.opConfObjList:
875 # opConfObj.makeXml(procUnitElement)
876
877 877 def readXml(self, upElement):
878 878
879 879 self.id = upElement.get('id')
880 880 self.name = upElement.get('name')
881 881 self.datatype = upElement.get('datatype')
882 882 self.inputId = upElement.get('inputId')
883 883
884 884 if self.ELEMENTNAME == "ReadUnit":
885 885 self.datatype = self.datatype.replace("Reader", "")
886 886
887 887 if self.inputId == 'None':
888 888 self.inputId = '0'
889 889
890 890 self.opConfObjList = []
891 891
892 892 opElementList = upElement.iter(OperationConf().getElementName())
893 893
894 894 for opElement in opElementList:
895 895 opConfObj = OperationConf()
896 896 opConfObj.readXml(opElement)
897 897 self.opConfObjList.append(opConfObj)
898 898
899 899 if opConfObj.name == 'run':
900 900 self.path = opConfObj.getParameterValue('path')
901 901 self.startDate = opConfObj.getParameterValue('startDate')
902 902 self.endDate = opConfObj.getParameterValue('endDate')
903 903 self.startTime = opConfObj.getParameterValue('startTime')
904 904 self.endTime = opConfObj.getParameterValue('endTime')
905 905
906 906 class Project():
907 907
908 908 id = None
909 909 name = None
910 910 description = None
911 911 filename = None
912 912
913 913 procUnitConfObjDict = None
914 914
915 915 ELEMENTNAME = 'Project'
916 916
917 917 plotterQueue = None
918 918
919 919 def __init__(self, plotter_queue=None):
920 920
921 921 self.id = None
922 922 self.name = None
923 923 self.description = None
924 924
925 925 self.plotterQueue = plotter_queue
926 926
927 927 self.procUnitConfObjDict = {}
928 928
929 929 def __getNewId(self):
930 930
931 931 idList = self.procUnitConfObjDict.keys()
932 932
933 933 id = int(self.id)*10
934 934
935 935 while True:
936 936 id += 1
937 937
938 938 if str(id) in idList:
939 939 continue
940 940
941 941 break
942 942
943 943 return str(id)
944 944
945 945 def getElementName(self):
946 946
947 947 return self.ELEMENTNAME
948 948
949 949 def getId(self):
950 950
951 951 return self.id
952 952
953 953 def updateId(self, new_id):
954 954
955 955 self.id = str(new_id)
956 956
957 957 keyList = self.procUnitConfObjDict.keys()
958 958 keyList.sort()
959 959
960 960 n = 1
961 961 newProcUnitConfObjDict = {}
962 962
963 963 for procKey in keyList:
964 964
965 965 procUnitConfObj = self.procUnitConfObjDict[procKey]
966 966 idProcUnit = str(int(self.id)*10 + n)
967 967 procUnitConfObj.updateId(idProcUnit, parentId = self.id)
968 968
969 969 newProcUnitConfObjDict[idProcUnit] = procUnitConfObj
970 970 n += 1
971 971
972 972 self.procUnitConfObjDict = newProcUnitConfObjDict
973 973
974 974 def setup(self, id, name, description):
975 975
976 976 self.id = str(id)
977 977 self.name = name
978 978 self.description = description
979 979
980 980 def update(self, name, description):
981 981
982 982 self.name = name
983 983 self.description = description
984 984
985 985 def addReadUnit(self, id=None, datatype=None, name=None, **kwargs):
986 986
987 987 if id is None:
988 988 idReadUnit = self.__getNewId()
989 989 else:
990 990 idReadUnit = str(id)
991 991
992 992 readUnitConfObj = ReadUnitConf()
993 993 readUnitConfObj.setup(idReadUnit, name, datatype, parentId=self.id, **kwargs)
994 994
995 995 self.procUnitConfObjDict[readUnitConfObj.getId()] = readUnitConfObj
996 996
997 997 return readUnitConfObj
998 998
999 999 def addProcUnit(self, inputId='0', datatype=None, name=None):
1000 1000
1001 1001 idProcUnit = self.__getNewId()
1002 1002
1003 1003 procUnitConfObj = ProcUnitConf()
1004 1004 procUnitConfObj.setup(idProcUnit, name, datatype, inputId, parentId=self.id)
1005 1005
1006 1006 self.procUnitConfObjDict[procUnitConfObj.getId()] = procUnitConfObj
1007 1007
1008 1008 return procUnitConfObj
1009 1009
1010 1010 def removeProcUnit(self, id):
1011 1011
1012 1012 if id in self.procUnitConfObjDict.keys():
1013 1013 self.procUnitConfObjDict.pop(id)
1014 1014
1015 1015 def getReadUnitId(self):
1016 1016
1017 1017 readUnitConfObj = self.getReadUnitObj()
1018 1018
1019 1019 return readUnitConfObj.id
1020 1020
1021 1021 def getReadUnitObj(self):
1022 1022
1023 1023 for obj in self.procUnitConfObjDict.values():
1024 1024 if obj.getElementName() == "ReadUnit":
1025 1025 return obj
1026 1026
1027 1027 return None
1028 1028
1029 1029 def getProcUnitObj(self, id=None, name=None):
1030 1030
1031 1031 if id != None:
1032 1032 return self.procUnitConfObjDict[id]
1033 1033
1034 1034 if name != None:
1035 1035 return self.getProcUnitObjByName(name)
1036 1036
1037 1037 return None
1038 1038
1039 1039 def getProcUnitObjByName(self, name):
1040 1040
1041 1041 for obj in self.procUnitConfObjDict.values():
1042 1042 if obj.name == name:
1043 1043 return obj
1044 1044
1045 1045 return None
1046 1046
1047 1047 def procUnitItems(self):
1048 1048
1049 1049 return self.procUnitConfObjDict.items()
1050 1050
1051 1051 def makeXml(self):
1052 1052
1053 1053 projectElement = Element('Project')
1054 1054 projectElement.set('id', str(self.id))
1055 1055 projectElement.set('name', self.name)
1056 1056 projectElement.set('description', self.description)
1057 1057
1058 1058 for procUnitConfObj in self.procUnitConfObjDict.values():
1059 1059 procUnitConfObj.makeXml(projectElement)
1060 1060
1061 1061 self.projectElement = projectElement
1062 1062
1063 1063 def writeXml(self, filename=None):
1064 1064
1065 1065 if filename == None:
1066 1066 if self.filename:
1067 1067 filename = self.filename
1068 1068 else:
1069 1069 filename = "schain.xml"
1070 1070
1071 1071 if not filename:
1072 1072 print "filename has not been defined. Use setFilename(filename) for do it."
1073 1073 return 0
1074 1074
1075 1075 abs_file = os.path.abspath(filename)
1076 1076
1077 1077 if not os.access(os.path.dirname(abs_file), os.W_OK):
1078 1078 print "No write permission on %s" %os.path.dirname(abs_file)
1079 1079 return 0
1080 1080
1081 1081 if os.path.isfile(abs_file) and not(os.access(abs_file, os.W_OK)):
1082 1082 print "File %s already exists and it could not be overwriten" %abs_file
1083 1083 return 0
1084 1084
1085 1085 self.makeXml()
1086 1086
1087 1087 ElementTree(self.projectElement).write(abs_file, method='xml')
1088 1088
1089 1089 self.filename = abs_file
1090 1090
1091 1091 return 1
1092 1092
1093 1093 def readXml(self, filename = None):
1094 1094
1095 1095 if not filename:
1096 1096 print "filename is not defined"
1097 1097 return 0
1098 1098
1099 1099 abs_file = os.path.abspath(filename)
1100 1100
1101 1101 if not os.path.isfile(abs_file):
1102 1102 print "%s file does not exist" %abs_file
1103 1103 return 0
1104 1104
1105 1105 self.projectElement = None
1106 1106 self.procUnitConfObjDict = {}
1107 1107
1108 1108 try:
1109 1109 self.projectElement = ElementTree().parse(abs_file)
1110 1110 except:
1111 1111 print "Error reading %s, verify file format" %filename
1112 1112 return 0
1113 1113
1114 1114 self.project = self.projectElement.tag
1115 1115
1116 1116 self.id = self.projectElement.get('id')
1117 1117 self.name = self.projectElement.get('name')
1118 self.description = self.projectElement.get('description')
1119
1118 self.description = self.projectElement.get('description')
1119
1120 1120 readUnitElementList = self.projectElement.iter(ReadUnitConf().getElementName())
1121 1121
1122 1122 for readUnitElement in readUnitElementList:
1123 1123 readUnitConfObj = ReadUnitConf()
1124 1124 readUnitConfObj.readXml(readUnitElement)
1125 1125
1126 1126 if readUnitConfObj.parentId == None:
1127 1127 readUnitConfObj.parentId = self.id
1128 1128
1129 1129 self.procUnitConfObjDict[readUnitConfObj.getId()] = readUnitConfObj
1130 1130
1131 1131 procUnitElementList = self.projectElement.iter(ProcUnitConf().getElementName())
1132 1132
1133 1133 for procUnitElement in procUnitElementList:
1134 1134 procUnitConfObj = ProcUnitConf()
1135 1135 procUnitConfObj.readXml(procUnitElement)
1136 1136
1137 1137 if procUnitConfObj.parentId == None:
1138 1138 procUnitConfObj.parentId = self.id
1139 1139
1140 1140 self.procUnitConfObjDict[procUnitConfObj.getId()] = procUnitConfObj
1141 1141
1142 1142 self.filename = abs_file
1143 1143
1144 1144 return 1
1145 1145
1146 1146 def printattr(self):
1147 1147
1148 1148 print "Project[%s]: name = %s, description = %s" %(self.id,
1149 self.name,
1150 self.description)
1151
1149 self.name,
1150 self.description)
1151
1152 1152 for procUnitConfObj in self.procUnitConfObjDict.values():
1153 1153 procUnitConfObj.printattr()
1154 1154
1155 1155 def createObjects(self):
1156 1156
1157 1157 for procUnitConfObj in self.procUnitConfObjDict.values():
1158 1158 procUnitConfObj.createObjects(self.plotterQueue)
1159 1159
1160 1160 def __connect(self, objIN, thisObj):
1161 1161
1162 1162 thisObj.setInput(objIN.getOutputObj())
1163 1163
1164 1164 def connectObjects(self):
1165 1165
1166 1166 for thisPUConfObj in self.procUnitConfObjDict.values():
1167 1167
1168 1168 inputId = thisPUConfObj.getInputId()
1169 1169
1170 1170 if int(inputId) == 0:
1171 1171 continue
1172 1172
1173 1173 #Get input object
1174 1174 puConfINObj = self.procUnitConfObjDict[inputId]
1175 1175 puObjIN = puConfINObj.getProcUnitObj()
1176 1176
1177 1177 #Get current object
1178 1178 thisPUObj = thisPUConfObj.getProcUnitObj()
1179 1179
1180 1180 self.__connect(puObjIN, thisPUObj)
1181 1181
1182 1182 def __handleError(self, procUnitConfObj, send_email=True):
1183 1183
1184 1184 import socket
1185 1185
1186 1186 err = traceback.format_exception(sys.exc_info()[0],
1187 sys.exc_info()[1],
1188 sys.exc_info()[2])
1189
1187 sys.exc_info()[1],
1188 sys.exc_info()[2])
1189
1190 1190 print "***** Error occurred in %s *****" %(procUnitConfObj.name)
1191 1191 print "***** %s" %err[-1]
1192 1192
1193 1193 message = "".join(err)
1194 1194
1195 1195 sys.stderr.write(message)
1196 1196
1197 1197 if not send_email:
1198 1198 return
1199 1199
1200 1200 subject = "SChain v%s: Error running %s\n" %(schainpy.__version__, procUnitConfObj.name)
1201 1201
1202 1202 subtitle = "%s: %s\n" %(procUnitConfObj.getElementName() ,procUnitConfObj.name)
1203 1203 subtitle += "Hostname: %s\n" %socket.gethostbyname(socket.gethostname())
1204 1204 subtitle += "Working directory: %s\n" %os.path.abspath("./")
1205 1205 subtitle += "Configuration file: %s\n" %self.filename
1206 1206 subtitle += "Time: %s\n" %str(datetime.datetime.now())
1207 1207
1208 1208 readUnitConfObj = self.getReadUnitObj()
1209 1209 if readUnitConfObj:
1210 1210 subtitle += "\nInput parameters:\n"
1211 1211 subtitle += "[Data path = %s]\n" %readUnitConfObj.path
1212 1212 subtitle += "[Data type = %s]\n" %readUnitConfObj.datatype
1213 1213 subtitle += "[Start date = %s]\n" %readUnitConfObj.startDate
1214 1214 subtitle += "[End date = %s]\n" %readUnitConfObj.endDate
1215 1215 subtitle += "[Start time = %s]\n" %readUnitConfObj.startTime
1216 1216 subtitle += "[End time = %s]\n" %readUnitConfObj.endTime
1217 1217
1218 1218 adminObj = schainpy.admin.SchainNotify()
1219 1219 adminObj.sendAlert(message=message,
1220 subject=subject,
1221 subtitle=subtitle,
1222 filename=self.filename)
1223
1220 subject=subject,
1221 subtitle=subtitle,
1222 filename=self.filename)
1223
1224 1224 def isPaused(self):
1225 1225 return 0
1226 1226
1227 1227 def isStopped(self):
1228 1228 return 0
1229 1229
1230 1230 def runController(self):
1231 1231 """
1232 1232 returns 0 when this process has been stopped, 1 otherwise
1233 1233 """
1234 1234
1235 1235 if self.isPaused():
1236 1236 print "Process suspended"
1237 1237
1238 1238 while True:
1239 1239 sleep(0.1)
1240 1240
1241 1241 if not self.isPaused():
1242 1242 break
1243 1243
1244 1244 if self.isStopped():
1245 1245 break
1246 1246
1247 1247 print "Process reinitialized"
1248 1248
1249 1249 if self.isStopped():
1250 1250 print "Process stopped"
1251 1251 return 0
1252 1252
1253 1253 return 1
1254 1254
1255 1255 def setFilename(self, filename):
1256 1256
1257 1257 self.filename = filename
1258 1258
1259 1259 def setPlotterQueue(self, plotter_queue):
1260 1260
1261 1261 raise NotImplementedError, "Use schainpy.controller_api.ControllerThread instead Project class"
1262 1262
1263 1263 def getPlotterQueue(self):
1264 1264
1265 1265 raise NotImplementedError, "Use schainpy.controller_api.ControllerThread instead Project class"
1266 1266
1267 1267 def useExternalPlotter(self):
1268 1268
1269 1269 raise NotImplementedError, "Use schainpy.controller_api.ControllerThread instead Project class"
1270 1270
1271 1271 def run(self):
1272 1272
1273 1273 print
1274 1274 print "*"*60
1275 1275 print " Starting SIGNAL CHAIN PROCESSING v%s " %schainpy.__version__
1276 1276 print "*"*60
1277 1277 print
1278 1278
1279 1279 keyList = self.procUnitConfObjDict.keys()
1280 1280 keyList.sort()
1281 1281
1282 1282 while(True):
1283 1283
1284 1284 is_ok = False
1285 1285
1286 1286 for procKey in keyList:
1287 1287 # print "Running the '%s' process with %s" %(procUnitConfObj.name, procUnitConfObj.id)
1288 1288
1289 1289 procUnitConfObj = self.procUnitConfObjDict[procKey]
1290 1290
1291 1291 try:
1292 1292 sts = procUnitConfObj.run()
1293 1293 is_ok = is_ok or sts
1294 1294 except KeyboardInterrupt:
1295 1295 is_ok = False
1296 1296 break
1297 1297 except ValueError, e:
1298 1298 sleep(0.5)
1299 1299 self.__handleError(procUnitConfObj, send_email=True)
1300 1300 is_ok = False
1301 1301 break
1302 1302 except:
1303 1303 sleep(0.5)
1304 1304 self.__handleError(procUnitConfObj)
1305 1305 is_ok = False
1306 1306 break
1307 1307
1308 1308 #If every process unit finished so end process
1309 1309 if not(is_ok):
1310 1310 # print "Every process unit have finished"
1311 1311 break
1312 1312
1313 1313 if not self.runController():
1314 1314 break
1315 1315
1316 1316 #Closing every process
1317 1317 for procKey in keyList:
1318 1318 procUnitConfObj = self.procUnitConfObjDict[procKey]
1319 1319 procUnitConfObj.close()
1320 1320
1321 1321 print "Process finished"
1322 1322
1323 1323 def start(self, filename=None):
1324 1324
1325 1325 self.writeXml(filename)
1326 1326 self.createObjects()
1327 1327 self.connectObjects()
1328 1328 self.run()
Show file before | Show file after | Hide comments
@@ -1,179 +1,179
1 1 import threading
2 2 from Queue import Queue
3 3
4 4 from schainpy.controller import Project
5 5 from schainpy.model.graphics.jroplotter import PlotManager
6 6
7 7 class ControllerThread(threading.Thread, Project):
8 8
9 9 def __init__(self, plotter_queue=None):
10 10
11 11 threading.Thread.__init__(self)
12 12 Project.__init__(self, plotter_queue)
13 13
14 14 self.setDaemon(True)
15 15
16 16 self.lock = threading.Lock()
17 17 self.control = {'stop':False, 'pause':False}
18 18
19 19 def __del__(self):
20 20
21 21 self.control['stop'] = True
22 22
23 23 def stop(self):
24 24
25 25 self.lock.acquire()
26 26
27 27 self.control['stop'] = True
28 28
29 29 self.lock.release()
30 30
31 31 def pause(self):
32 32
33 33 self.lock.acquire()
34 34
35 35 self.control['pause'] = not(self.control['pause'])
36 36 paused = self.control['pause']
37 37
38 38 self.lock.release()
39 39
40 40 return paused
41 41
42 42 def isPaused(self):
43 43
44 44 self.lock.acquire()
45 45 paused = self.control['pause']
46 46 self.lock.release()
47 47
48 48 return paused
49 49
50 50 def isStopped(self):
51 51
52 52 self.lock.acquire()
53 53 stopped = self.control['stop']
54 54 self.lock.release()
55 55
56 56 return stopped
57 57
58 58 def run(self):
59 59 self.control['stop'] = False
60 60 self.control['pause'] = False
61 61
62 62 self.writeXml()
63 63
64 64 self.createObjects()
65 65 self.connectObjects()
66 66 Project.run(self)
67 67
68 68 def isRunning(self):
69 69
70 70 return self.is_alive()
71 71
72 72 def isFinished(self):
73 73
74 74 return not self.is_alive()
75 75
76 76 def setPlotters(self):
77 77
78 78 plotterList = PlotManager.plotterList
79 79
80 80 for thisPUConfObj in self.procUnitConfObjDict.values():
81 81
82 82 inputId = thisPUConfObj.getInputId()
83 83
84 84 if int(inputId) == 0:
85 85 continue
86 86
87 87 for thisOpObj in thisPUConfObj.getOperationObjList():
88 88
89 89 if thisOpObj.type == "self":
90 90 continue
91 91
92 92 if thisOpObj.name in plotterList:
93 thisOpObj.type = "plotter"
93 thisOpObj.type = "other"
94 94
95 95 def setPlotterQueue(self, plotter_queue):
96 96
97 97 self.plotterQueue = plotter_queue
98 98
99 99 def getPlotterQueue(self):
100 100
101 101 return self.plotterQueue
102 102
103 103 def useExternalPlotter(self):
104 104
105 105 self.plotterQueue = Queue(10)
106 106 self.setPlotters()
107 107
108 108 plotManagerObj = PlotManager(self.plotterQueue)
109 109 plotManagerObj.setController(self)
110 110
111 111 return plotManagerObj
112 112
113 113 # from PyQt4 import QtCore
114 114 # from PyQt4.QtCore import SIGNAL
115 115 #
116 116 # class ControllerQThread(QtCore.QThread, Project):
117 117 #
118 118 # def __init__(self, filename):
119 119 #
120 120 # QtCore.QThread.__init__(self)
121 121 # Project.__init__(self)
122 122 #
123 123 # self.filename = filename
124 124 #
125 125 # self.lock = threading.Lock()
126 126 # self.control = {'stop':False, 'pause':False}
127 127 #
128 128 # def __del__(self):
129 129 #
130 130 # self.control['stop'] = True
131 131 # self.wait()
132 132 #
133 133 # def stop(self):
134 134 #
135 135 # self.lock.acquire()
136 136 #
137 137 # self.control['stop'] = True
138 138 #
139 139 # self.lock.release()
140 140 #
141 141 # def pause(self):
142 142 #
143 143 # self.lock.acquire()
144 144 #
145 145 # self.control['pause'] = not(self.control['pause'])
146 146 # paused = self.control['pause']
147 147 #
148 148 # self.lock.release()
149 149 #
150 150 # return paused
151 151 #
152 152 # def isPaused(self):
153 153 #
154 154 # self.lock.acquire()
155 155 # paused = self.control['pause']
156 156 # self.lock.release()
157 157 #
158 158 # return paused
159 159 #
160 160 # def isStopped(self):
161 161 #
162 162 # self.lock.acquire()
163 163 # stopped = self.control['stop']
164 164 # self.lock.release()
165 165 #
166 166 # return stopped
167 167 #
168 168 # def run(self):
169 169 #
170 170 # self.control['stop'] = False
171 171 # self.control['pause'] = False
172 172 #
173 173 # self.readXml(self.filename)
174 174 # self.createObjects()
175 175 # self.connectObjects()
176 176 # self.emit( SIGNAL( "jobStarted( PyQt_PyObject )" ), 1)
177 177 # Project.run(self)
178 178 # self.emit( SIGNAL( "jobFinished( PyQt_PyObject )" ), 1)
179 179 #
Show file before | Show file after | Hide comments
@@ -1,1007 +1,1009
1 1 '''
2 2 @author: Daniel Suarez
3 3 '''
4 4 import os
5 5 import glob
6 6 import ftplib
7 7
8 8 try:
9 9 import paramiko
10 10 import scp
11 11 except:
12 12 print "You should install paramiko and scp libraries \nif you want to use SSH protocol to upload files to the server"
13 13
14 14 import time
15 15
16 16 import threading
17 17 Thread = threading.Thread
18 18
19 19 # try:
20 20 # from gevent import sleep
21 21 # except:
22 22 from time import sleep
23 23
24 24 from schainpy.model.proc.jroproc_base import ProcessingUnit, Operation
25 25
26 26 class Remote(Thread):
27 27 """
28 28 Remote is a parent class used to define the behaviour of FTP and SSH class. These clases are
29 29 used to upload or download files remotely.
30 30
31 31 Non-standard Python modules used:
32 32 None
33 33
34 34 Written by:
35 35
36 36 "Miguel Urco":mailto:miguel.urco@jro.igp.gob.pe Jun. 03, 2015
37 37
38 38 """
39 39
40 40 server = None
41 41 username = None
42 42 password = None
43 43 remotefolder = None
44 44
45 45 period = 60
46 46 fileList = []
47 47 bussy = False
48 48
49 49 def __init__(self, server, username, password, remotefolder, period=60):
50 50
51 51 Thread.__init__(self)
52 52
53 53 self.setDaemon(True)
54 54
55 55 self.status = 0
56 56
57 57 self.__server = server
58 58 self.__username = username
59 59 self.__password = password
60 60 self.__remotefolder = remotefolder
61 61
62 62 self.period = period
63 63
64 64 self.fileList = []
65 65 self.bussy = False
66 66
67 67 self.stopFlag = False
68 68
69 69 print "[Remote Server] Opening server: %s" %self.__server
70 70 if self.open(self.__server, self.__username, self.__password, self.__remotefolder):
71 71 print "[Remote Server] %s server was opened successfully" %self.__server
72 72
73 73 self.close()
74 74
75 75 self.mutex = threading.Lock()
76 76
77 77 def stop(self):
78 78
79 79 self.stopFlag = True
80 80 self.join(10)
81 81
82 82 def open(self):
83 83 """
84 84 Connect to server and create a connection class (FTP or SSH) to remote server.
85 85 """
86 86 raise NotImplementedError, "Implement this method in child class"
87 87
88 88 def close(self):
89 89 """
90 90 Close connection to server
91 91 """
92 92 raise NotImplementedError, "Implement this method in child class"
93 93
94 94 def mkdir(self, remotefolder):
95 95 """
96 96 Create a folder remotely
97 97 """
98 98 raise NotImplementedError, "Implement this method in child class"
99 99
100 100 def cd(self, remotefolder):
101 101 """
102 102 Change working directory in remote server
103 103 """
104 104 raise NotImplementedError, "Implement this method in child class"
105 105
106 106 def download(self, filename, localfolder=None):
107 107 """
108 108 Download a file from server to local host
109 109 """
110 110 raise NotImplementedError, "Implement this method in child class"
111 111
112 112 def sendFile(self, fullfilename):
113 113 """
114 114 sendFile method is used to upload a local file to the current directory in remote server
115 115
116 116 Inputs:
117 117 fullfilename - full path name of local file to store in remote directory
118 118
119 119 Returns:
120 120 0 in error case else 1
121 121 """
122 122 raise NotImplementedError, "Implement this method in child class"
123 123
124 124 def upload(self, fullfilename, remotefolder=None):
125 125 """
126 126 upload method is used to upload a local file to remote directory. This method changes
127 127 working directory before sending a file.
128 128
129 129 Inputs:
130 130 fullfilename - full path name of local file to store in remote directory
131 131
132 132 remotefolder - remote directory
133 133
134 134 Returns:
135 135 0 in error case else 1
136 136 """
137 137 print "[Remote Server] Uploading %s to %s:%s" %(fullfilename, self.server, self.remotefolder)
138 138
139 139 if not self.status:
140 140 return 0
141 141
142 142 if remotefolder == None:
143 143 remotefolder = self.remotefolder
144 144
145 145 if not self.cd(remotefolder):
146 146 return 0
147 147
148 148 if not self.sendFile(fullfilename):
149 149 print "[Remote Server] Error uploading file %s" %fullfilename
150 150 return 0
151 151
152 152 print "[Remote Server] upload finished successfully"
153 153
154 154 return 1
155 155
156 156 def delete(self, filename):
157 157 """
158 158 Remove a file from remote server
159 159 """
160 160 pass
161 161
162 162 def updateFileList(self, fileList):
163 163 """
164 164 Remove a file from remote server
165 165 """
166 166
167 167 if fileList == self.fileList:
168 168 return 0
169 169
170 170 self.mutex.acquire()
171 171 # init = time.time()
172 172 #
173 173 # while(self.bussy):
174 174 # sleep(0.1)
175 175 # if time.time() - init > 2*self.period:
176 176 # return 0
177 177
178 178 self.fileList = fileList
179 179 self.mutex.release()
180 180 return 1
181 181
182 182 def run(self):
183 183
184 184 if not self.status:
185 185 print "Finishing FTP service"
186 186 return
187 187
188 188 if not self.cd(self.remotefolder):
189 189 raise ValueError, "Could not access to the new remote directory: %s" %self.remotefolder
190 190
191 191 while True:
192 192
193 193 for i in range(self.period):
194 194 if self.stopFlag:
195 195 break
196 196 sleep(1)
197 197
198 198 if self.stopFlag:
199 199 break
200 200
201 201 # self.bussy = True
202 202 self.mutex.acquire()
203 203
204 204 print "[Remote Server] Opening %s" %self.__server
205 205 if not self.open(self.__server, self.__username, self.__password, self.__remotefolder):
206 206 self.mutex.release()
207 207 continue
208 208
209 209 for thisFile in self.fileList:
210 210 self.upload(thisFile, self.remotefolder)
211 211
212 212 print "[Remote Server] Closing %s" %self.__server
213 213 self.close()
214 214
215 215 self.mutex.release()
216 216 # self.bussy = False
217 217
218 218 print "[Remote Server] Thread stopped successfully"
219 219
220 220 class FTPClient(Remote):
221 221
222 222 __ftpClientObj = None
223 223
224 224 def __init__(self, server, username, password, remotefolder, period=60):
225 225 """
226 226 """
227 227 Remote.__init__(self, server, username, password, remotefolder, period)
228 228
229 229 def open(self, server, username, password, remotefolder):
230 230
231 231 """
232 232 This method is used to set FTP parameters and establish a connection to remote server
233 233
234 234 Inputs:
235 235 server - remote server IP Address
236 236
237 237 username - remote server Username
238 238
239 239 password - remote server password
240 240
241 241 remotefolder - remote server current working directory
242 242
243 243 Return:
244 244 Boolean - Returns 1 if a connection has been established, 0 otherwise
245 245
246 246 Affects:
247 247 self.status - in case of error or fail connection this parameter is set to 0 else 1
248 248
249 249 """
250 250
251 251 if server == None:
252 252 raise ValueError, "FTP server should be defined"
253 253
254 254 if username == None:
255 255 raise ValueError, "FTP username should be defined"
256 256
257 257 if password == None:
258 258 raise ValueError, "FTP password should be defined"
259 259
260 260 if remotefolder == None:
261 261 raise ValueError, "FTP remote folder should be defined"
262 262
263 263 try:
264 264 ftpClientObj = ftplib.FTP(server)
265 265 except ftplib.all_errors, e:
266 266 print "[FTP Server]: FTP server connection fail: %s" %server
267 267 print "[FTP Server]:", e
268 268 self.status = 0
269 269 return 0
270 270
271 271 try:
272 272 ftpClientObj.login(username, password)
273 273 except ftplib.all_errors:
274 274 print "[FTP Server]: FTP username or password are incorrect"
275 275 self.status = 0
276 276 return 0
277 277
278 278 if remotefolder == None:
279 279 remotefolder = ftpClientObj.pwd()
280 280 else:
281 281 try:
282 282 ftpClientObj.cwd(remotefolder)
283 283 except ftplib.all_errors:
284 284 print "[FTP Server]: FTP remote folder is invalid: %s" %remotefolder
285 285 remotefolder = ftpClientObj.pwd()
286 286
287 287 self.server = server
288 288 self.username = username
289 289 self.password = password
290 290 self.remotefolder = remotefolder
291 291 self.__ftpClientObj = ftpClientObj
292 292 self.status = 1
293 293
294 294 return 1
295 295
296 296 def close(self):
297 297 """
298 298 Close connection to remote server
299 299 """
300 300 if not self.status:
301 301 return 0
302 302
303 303 self.__ftpClientObj.close()
304 304
305 305 def mkdir(self, remotefolder):
306 306 """
307 307 mkdir is used to make a new directory in remote server
308 308
309 309 Input:
310 310 remotefolder - directory name
311 311
312 312 Return:
313 313 0 in error case else 1
314 314 """
315 315 if not self.status:
316 316 return 0
317 317
318 318 try:
319 319 self.__ftpClientObj.mkd(dirname)
320 320 except ftplib.all_errors:
321 321 print "[FTP Server]: Error creating remote folder: %s" %remotefolder
322 322 return 0
323 323
324 324 return 1
325 325
326 326 def cd(self, remotefolder):
327 327 """
328 328 cd is used to change remote working directory on server
329 329
330 330 Input:
331 331 remotefolder - current working directory
332 332
333 333 Affects:
334 334 self.remotefolder
335 335
336 336 Return:
337 337 0 in case of error else 1
338 338 """
339 339 if not self.status:
340 340 return 0
341 341
342 342 if remotefolder == self.remotefolder:
343 343 return 1
344 344
345 345 try:
346 346 self.__ftpClientObj.cwd(remotefolder)
347 347 except ftplib.all_errors:
348 348 print '[FTP Server]: Error changing to %s' %remotefolder
349 349 print '[FTP Server]: Trying to create remote folder'
350 350
351 351 if not self.mkdir(remotefolder):
352 352 print '[FTP Server]: Remote folder could not be created'
353 353 return 0
354 354
355 355 try:
356 356 self.__ftpClientObj.cwd(remotefolder)
357 357 except ftplib.all_errors:
358 358 return 0
359 359
360 360 self.remotefolder = remotefolder
361 361
362 362 return 1
363 363
364 364 def sendFile(self, fullfilename):
365 365
366 366 if not self.status:
367 367 return 0
368 368
369 369 fp = open(fullfilename, 'rb')
370 370
371 371 filename = os.path.basename(fullfilename)
372 372
373 373 command = "STOR %s" %filename
374 374
375 375 try:
376 376 self.__ftpClientObj.storbinary(command, fp)
377 377 except ftplib.all_errors, e:
378 378 print "[FTP Server]:", e
379 379 return 0
380 380
381 381 try:
382 382 self.__ftpClientObj.sendcmd('SITE CHMOD 755 ' + filename)
383 383 except ftplib.all_errors, e:
384 384 print "[FTP Server]:", e
385 385
386 386 fp.close()
387 387
388 388 return 1
389 389
390 390 class SSHClient(Remote):
391 391
392 392 __sshClientObj = None
393 393 __scpClientObj = None
394 394
395 395 def __init__(self, server, username, password, remotefolder, period=60):
396 396 """
397 397 """
398 398 Remote.__init__(self, server, username, password, remotefolder, period)
399 399
400 400 def open(self, server, username, password, remotefolder, port=22):
401 401
402 402 """
403 403 This method is used to set SSH parameters and establish a connection to a remote server
404 404
405 405 Inputs:
406 406 server - remote server IP Address
407 407
408 408 username - remote server Username
409 409
410 410 password - remote server password
411 411
412 412 remotefolder - remote server current working directory
413 413
414 414 Return: void
415 415
416 416 Affects:
417 417 self.status - in case of error or fail connection this parameter is set to 0 else 1
418 418
419 419 """
420 420 import socket
421 421
422 422 if server == None:
423 423 raise ValueError, "SSH server should be defined"
424 424
425 425 if username == None:
426 426 raise ValueError, "SSH username should be defined"
427 427
428 428 if password == None:
429 429 raise ValueError, "SSH password should be defined"
430 430
431 431 if remotefolder == None:
432 432 raise ValueError, "SSH remote folder should be defined"
433 433
434 434 sshClientObj = paramiko.SSHClient()
435 435
436 436 sshClientObj.load_system_host_keys()
437 437 sshClientObj.set_missing_host_key_policy(paramiko.WarningPolicy())
438 438
439 439 self.status = 0
440 440 try:
441 441 sshClientObj.connect(server, username=username, password=password, port=port)
442 442 except paramiko.AuthenticationException, e:
443 443 # print "SSH username or password are incorrect: %s"
444 444 print "[SSH Server]:", e
445 445 return 0
446 446 except SSHException, e:
447 447 print "[SSH Server]:", e
448 448 return 0
449 449 except socket.error:
450 450 self.status = 0
451 451 print "[SSH Server]:", e
452 452 return 0
453 453
454 454 self.status = 1
455 455 scpClientObj = scp.SCPClient(sshClientObj.get_transport(), socket_timeout=30)
456 456
457 457 if remotefolder == None:
458 458 remotefolder = self.pwd()
459 459
460 460 self.server = server
461 461 self.username = username
462 462 self.password = password
463 463 self.__sshClientObj = sshClientObj
464 464 self.__scpClientObj = scpClientObj
465 465 self.status = 1
466 466
467 467 if not self.cd(remotefolder):
468 468 raise ValueError, "[SSH Server]: Could not access to remote folder: %s" %remotefolder
469 469 return 0
470 470
471 471 self.remotefolder = remotefolder
472 472
473 473 return 1
474 474
475 475 def close(self):
476 476 """
477 477 Close connection to remote server
478 478 """
479 479 if not self.status:
480 480 return 0
481 481
482 482 self.__scpClientObj.close()
483 483 self.__sshClientObj.close()
484 484
485 485 def __execute(self, command):
486 486 """
487 487 __execute a command on remote server
488 488
489 489 Input:
490 490 command - Exmaple 'ls -l'
491 491
492 492 Return:
493 493 0 in error case else 1
494 494 """
495 495 if not self.status:
496 496 return 0
497 497
498 498 stdin, stdout, stderr = self.__sshClientObj.exec_command(command)
499 499
500 500 result = stderr.readlines()
501 501 if len(result) > 1:
502 502 return 0
503 503
504 504 result = stdout.readlines()
505 505 if len(result) > 1:
506 506 return result[0][:-1]
507 507
508 508 return 1
509 509
510 510 def mkdir(self, remotefolder):
511 511 """
512 512 mkdir is used to make a new directory in remote server
513 513
514 514 Input:
515 515 remotefolder - directory name
516 516
517 517 Return:
518 518 0 in error case else 1
519 519 """
520 520
521 521 command = 'mkdir %s' %remotefolder
522 522
523 523 return self.__execute(command)
524 524
525 525 def pwd(self):
526 526
527 527 command = 'pwd'
528 528
529 529 return self.__execute(command)
530 530
531 531 def cd(self, remotefolder):
532 532 """
533 533 cd is used to change remote working directory on server
534 534
535 535 Input:
536 536 remotefolder - current working directory
537 537
538 538 Affects:
539 539 self.remotefolder
540 540
541 541 Return:
542 542 0 in case of error else 1
543 543 """
544 544 if not self.status:
545 545 return 0
546 546
547 547 if remotefolder == self.remotefolder:
548 548 return 1
549 549
550 550 chk_command = "cd %s; pwd" %remotefolder
551 551 mkdir_command = "mkdir %s" %remotefolder
552 552
553 553 if not self.__execute(chk_command):
554 554 if not self.__execute(mkdir_command):
555 555 self.remotefolder = None
556 556 return 0
557 557
558 558 self.remotefolder = remotefolder
559 559
560 560 return 1
561 561
562 562 def sendFile(self, fullfilename):
563 563
564 564 if not self.status:
565 565 return 0
566 566
567 567 try:
568 568 self.__scpClientObj.put(fullfilename, remote_path=self.remotefolder)
569 569 except scp.ScpError, e:
570 570 print "[SSH Server]", str(e)
571 571 return 0
572 572
573 573 remotefile = os.path.join(self.remotefolder, os.path.split(fullfilename)[-1])
574 574 command = 'chmod 775 %s' %remotefile
575 575
576 576 return self.__execute(command)
577 577
578 578 class SendToServer(ProcessingUnit):
579 579
580 580 def __init__(self, **kwargs):
581 581
582 582 ProcessingUnit.__init__(self, **kwargs)
583 583
584 584 self.isConfig = False
585 585 self.clientObj = None
586 586
587 587 def setup(self, server, username, password, remotefolder, localfolder, ext='.png', period=60, protocol='ftp', **kwargs):
588 588
589 589 self.clientObj = None
590 590 self.localfolder = localfolder
591 591 self.ext = ext
592 592 self.period = period
593 593
594 594 if str.lower(protocol) == 'ftp':
595 595 self.clientObj = FTPClient(server, username, password, remotefolder, period)
596 596
597 597 if str.lower(protocol) == 'ssh':
598 598 self.clientObj = SSHClient(server, username, password, remotefolder, period)
599 599
600 600 if not self.clientObj:
601 601 raise ValueError, "%s has been chosen as remote access protocol but it is not valid" %protocol
602 602
603 603 self.clientObj.start()
604 604
605 605 def findFiles(self):
606 606
607 607 if not type(self.localfolder) == list:
608 608 folderList = [self.localfolder]
609 609 else:
610 610 folderList = self.localfolder
611 611
612 612 #Remove duplicate items
613 613 folderList = list(set(folderList))
614 614
615 615 fullfilenameList = []
616 616
617 617 for thisFolder in folderList:
618 618
619 619 print "[Remote Server]: Searching files on %s" %thisFolder
620 620
621 621 filenameList = glob.glob1(thisFolder, '*%s' %self.ext)
622 622
623 623 if len(filenameList) < 1:
624 624 continue
625 625
626 626 for thisFile in filenameList:
627 627 fullfilename = os.path.join(thisFolder, thisFile)
628 628
629 629 if fullfilename in fullfilenameList:
630 630 continue
631 631
632 632 #Only files modified in the last 30 minutes are considered
633 633 if os.path.getmtime(fullfilename) < time.time() - 30*60:
634 634 continue
635 635
636 636 fullfilenameList.append(fullfilename)
637 637
638 638 return fullfilenameList
639 639
640 640 def run(self, **kwargs):
641
642 641 if not self.isConfig:
643 642 self.init = time.time()
644 643 self.setup(**kwargs)
645 644 self.isConfig = True
646
645
646 if not self.clientObj.is_alive():
647 print "[Remote Server]: Restarting connection "
648 self.setup(**kwargs)
649
647 650 if time.time() - self.init >= self.period:
648 651 fullfilenameList = self.findFiles()
649 652
650 653 if self.clientObj.updateFileList(fullfilenameList):
651 654 print "[Remote Server]: Sending the next files ", str(fullfilenameList)
652
653 655 self.init = time.time()
654 656
655 657 def close(self):
656 658 print "[Remote Server] Stopping thread"
657 659 self.clientObj.stop()
658 660
659 661
660 662 class FTP(object):
661 663 """
662 664 Ftp is a public class used to define custom File Transfer Protocol from "ftplib" python module
663 665
664 666 Non-standard Python modules used: None
665 667
666 668 Written by "Daniel Suarez":mailto:daniel.suarez@jro.igp.gob.pe Oct. 26, 2010
667 669 """
668 670
669 671 def __init__(self,server = None, username=None, password=None, remotefolder=None):
670 672 """
671 673 This method is used to setting parameters for FTP and establishing connection to remote server
672 674
673 675 Inputs:
674 676 server - remote server IP Address
675 677
676 678 username - remote server Username
677 679
678 680 password - remote server password
679 681
680 682 remotefolder - remote server current working directory
681 683
682 684 Return: void
683 685
684 686 Affects:
685 687 self.status - in Error Case or Connection Failed this parameter is set to 1 else 0
686 688
687 689 self.folderList - sub-folder list of remote folder
688 690
689 691 self.fileList - file list of remote folder
690 692
691 693
692 694 """
693 695
694 696 if ((server == None) and (username==None) and (password==None) and (remotefolder==None)):
695 697 server, username, password, remotefolder = self.parmsByDefault()
696 698
697 699 self.server = server
698 700 self.username = username
699 701 self.password = password
700 702 self.remotefolder = remotefolder
701 703 self.file = None
702 704 self.ftp = None
703 705 self.status = 0
704 706
705 707 try:
706 708 self.ftp = ftplib.FTP(self.server)
707 709 self.ftp.login(self.username,self.password)
708 710 self.ftp.cwd(self.remotefolder)
709 711 # print 'Connect to FTP Server: Successfully'
710 712
711 713 except ftplib.all_errors:
712 714 print 'Error FTP Service'
713 715 self.status = 1
714 716 return
715 717
716 718
717 719
718 720 self.dirList = []
719 721
720 722 try:
721 723 self.dirList = self.ftp.nlst()
722 724
723 725 except ftplib.error_perm, resp:
724 726 if str(resp) == "550 No files found":
725 727 print "no files in this directory"
726 728 self.status = 1
727 729 return
728 730
729 731 except ftplib.all_errors:
730 732 print 'Error Displaying Dir-Files'
731 733 self.status = 1
732 734 return
733 735
734 736 self.fileList = []
735 737 self.folderList = []
736 738 #only for test
737 739 for f in self.dirList:
738 740 name, ext = os.path.splitext(f)
739 741 if ext != '':
740 742 self.fileList.append(f)
741 743 # print 'filename: %s - size: %d'%(f,self.ftp.size(f))
742 744
743 745 def parmsByDefault(self):
744 746 server = 'jro-app.igp.gob.pe'
745 747 username = 'wmaster'
746 748 password = 'mst2010vhf'
747 749 remotefolder = '/home/wmaster/graficos'
748 750
749 751 return server, username, password, remotefolder
750 752
751 753
752 754 def mkd(self,dirname):
753 755 """
754 756 mkd is used to make directory in remote server
755 757
756 758 Input:
757 759 dirname - directory name
758 760
759 761 Return:
760 762 1 in error case else 0
761 763 """
762 764 try:
763 765 self.ftp.mkd(dirname)
764 766 except:
765 767 print 'Error creating remote folder:%s'%dirname
766 768 return 1
767 769
768 770 return 0
769 771
770 772
771 773 def delete(self,filename):
772 774 """
773 775 delete is used to delete file in current working directory of remote server
774 776
775 777 Input:
776 778 filename - filename to delete in remote folder
777 779
778 780 Return:
779 781 1 in error case else 0
780 782 """
781 783
782 784 try:
783 785 self.ftp.delete(filename)
784 786 except:
785 787 print 'Error deleting remote file:%s'%filename
786 788 return 1
787 789
788 790 return 0
789 791
790 792 def download(self,filename,localfolder):
791 793 """
792 794 download is used to downloading file from remote folder into local folder
793 795
794 796 Inputs:
795 797 filename - filename to donwload
796 798
797 799 localfolder - directory local to store filename
798 800
799 801 Returns:
800 802 self.status - 1 in error case else 0
801 803 """
802 804
803 805 self.status = 0
804 806
805 807
806 808 if not(filename in self.fileList):
807 809 print 'filename:%s not exists'%filename
808 810 self.status = 1
809 811 return self.status
810 812
811 813 newfilename = os.path.join(localfolder,filename)
812 814
813 815 self.file = open(newfilename, 'wb')
814 816
815 817 try:
816 818 print 'Download: ' + filename
817 819 self.ftp.retrbinary('RETR ' + filename, self.__handleDownload)
818 820 print 'Download Complete'
819 821 except ftplib.all_errors:
820 822 print 'Error Downloading ' + filename
821 823 self.status = 1
822 824 return self.status
823 825
824 826 self.file.close()
825 827
826 828 return self.status
827 829
828 830
829 831 def __handleDownload(self,block):
830 832 """
831 833 __handleDownload is used to handle writing file
832 834 """
833 835 self.file.write(block)
834 836
835 837
836 838 def upload(self,filename,remotefolder=None):
837 839 """
838 840 upload is used to uploading local file to remote directory
839 841
840 842 Inputs:
841 843 filename - full path name of local file to store in remote directory
842 844
843 845 remotefolder - remote directory
844 846
845 847 Returns:
846 848 self.status - 1 in error case else 0
847 849 """
848 850
849 851 if remotefolder == None:
850 852 remotefolder = self.remotefolder
851 853
852 854 self.status = 0
853 855
854 856 try:
855 857 self.ftp.cwd(remotefolder)
856 858
857 859 self.file = open(filename, 'rb')
858 860
859 861 (head, tail) = os.path.split(filename)
860 862
861 863 command = "STOR " + tail
862 864
863 865 print 'Uploading: ' + tail
864 866 self.ftp.storbinary(command, self.file)
865 867 print 'Upload Completed'
866 868
867 869 except ftplib.all_errors:
868 870 print 'Error Uploading ' + tail
869 871 self.status = 1
870 872 return self.status
871 873
872 874 self.file.close()
873 875
874 876 #back to initial directory in __init__()
875 877 self.ftp.cwd(self.remotefolder)
876 878
877 879 return self.status
878 880
879 881
880 882 def dir(self,remotefolder):
881 883 """
882 884 dir is used to change working directory of remote server and get folder and file list
883 885
884 886 Input:
885 887 remotefolder - current working directory
886 888
887 889 Affects:
888 890 self.fileList - file list of working directory
889 891
890 892 Return:
891 893 infoList - list with filenames and size of file in bytes
892 894
893 895 self.folderList - folder list
894 896 """
895 897
896 898 self.remotefolder = remotefolder
897 899 print 'Change to ' + self.remotefolder
898 900 try:
899 901 self.ftp.cwd(remotefolder)
900 902 except ftplib.all_errors:
901 903 print 'Error Change to ' + self.remotefolder
902 904 infoList = None
903 905 self.folderList = None
904 906 return infoList,self.folderList
905 907
906 908 self.dirList = []
907 909
908 910 try:
909 911 self.dirList = self.ftp.nlst()
910 912
911 913 except ftplib.error_perm, resp:
912 914 if str(resp) == "550 No files found":
913 915 print "no files in this directory"
914 916 infoList = None
915 917 self.folderList = None
916 918 return infoList,self.folderList
917 919 except ftplib.all_errors:
918 920 print 'Error Displaying Dir-Files'
919 921 infoList = None
920 922 self.folderList = None
921 923 return infoList,self.folderList
922 924
923 925 infoList = []
924 926 self.fileList = []
925 927 self.folderList = []
926 928 for f in self.dirList:
927 929 name,ext = os.path.splitext(f)
928 930 if ext != '':
929 931 self.fileList.append(f)
930 932 value = (f,self.ftp.size(f))
931 933 infoList.append(value)
932 934
933 935 if ext == '':
934 936 self.folderList.append(f)
935 937
936 938 return infoList,self.folderList
937 939
938 940
939 941 def close(self):
940 942 """
941 943 close is used to close and end FTP connection
942 944
943 945 Inputs: None
944 946
945 947 Return: void
946 948
947 949 """
948 950 self.ftp.close()
949 951
950 952 class SendByFTP(Operation):
951 953
952 954 def __init__(self, **kwargs):
953 955 Operation.__init__(self, **kwargs)
954 956 self.status = 1
955 957 self.counter = 0
956 958
957 959 def error_print(self, ValueError):
958 960
959 961 print ValueError, 'Error FTP'
960 962 print "don't worry the program is running..."
961 963
962 964 def worker_ftp(self, server, username, password, remotefolder, filenameList):
963 965
964 966 self.ftpClientObj = FTP(server, username, password, remotefolder)
965 967 for filename in filenameList:
966 968 self.ftpClientObj.upload(filename)
967 969 self.ftpClientObj.close()
968 970
969 971 def ftp_thread(self, server, username, password, remotefolder):
970 972 if not(self.status):
971 973 return
972 974
973 975 import multiprocessing
974 976
975 977 p = multiprocessing.Process(target=self.worker_ftp, args=(server, username, password, remotefolder, self.filenameList,))
976 978 p.start()
977 979
978 980 p.join(3)
979 981
980 982 if p.is_alive():
981 983 p.terminate()
982 984 p.join()
983 985 print 'killing ftp process...'
984 986 self.status = 0
985 987 return
986 988
987 989 self.status = 1
988 990 return
989 991
990 992 def filterByExt(self, ext, localfolder):
991 993 fnameList = glob.glob1(localfolder,ext)
992 994 self.filenameList = [os.path.join(localfolder,x) for x in fnameList]
993 995
994 996 if len(self.filenameList) == 0:
995 997 self.status = 0
996 998
997 999 def run(self, dataOut, ext, localfolder, remotefolder, server, username, password, period=1):
998 1000
999 1001 self.counter += 1
1000 1002 if self.counter >= period:
1001 1003 self.filterByExt(ext, localfolder)
1002 1004
1003 1005 self.ftp_thread(server, username, password, remotefolder)
1004 1006
1005 1007 self.counter = 0
1006 1008
1007 1009 self.status = 1
General Comments 0
You need to be logged in to leave comments. Login now