ViewVC Help
View File | Revision Log | Show Annotations | Root Listing
root/cvsroot/COMP/CRAB/python/crab.py
Revision: 1.76
Committed: Wed Jul 19 10:02:10 2006 UTC (18 years, 9 months ago) by fanzago
Content type: text/x-python
Branch: MAIN
CVS Tags: HEAD_20092006
Changes since 1.75: +2 -2 lines
Log Message: 
test_tag

File Contents

# User Rev Content
1 elmer 1.65 #!/usr/bin/env python
2 nsmirnov 1.1 from crab_help import *
3     from crab_util import *
4     from crab_exceptions import *
5     from crab_logger import Logger
6     from WorkSpace import WorkSpace
7     from JobDB import JobDB
8 nsmirnov 1.3 from JobList import JobList
9 nsmirnov 1.1 from Creator import Creator
10     from Submitter import Submitter
11 slacapra 1.8 from Checker import Checker
12 slacapra 1.9 from PostMortem import PostMortem
13     from Status import Status
14 spiga 1.16 from StatusBoss import StatusBoss
15 corvo 1.42 from ApmonIf import ApmonIf
16 slacapra 1.36 from Cleaner import Cleaner
17 nsmirnov 1.1 import common
18 spiga 1.13 import Statistic
19 nsmirnov 1.1
20     import sys, os, time, string
21    
22     ###########################################################################
23     class Crab:
24 nsmirnov 1.3 def __init__(self, opts):
25 fanzago 1.76 ## test_tag
26 nsmirnov 1.1 # The order of main_actions is important !
27 slacapra 1.28 self.main_actions = [ '-create', '-submit' ]
28 slacapra 1.70 self.aux_actions = [ '-list', '-kill', '-status', '-getoutput','-get',
29 slacapra 1.67 '-resubmit' , '-cancelAndResubmit', '-testJdl', '-postMortem', '-clean',
30     '-printId' ]
31 nsmirnov 1.1
32     # Dictionary of actions, e.g. '-create' -> object of class Creator
33     self.actions = {}
34 fanzago 1.76
35 nsmirnov 1.1 # Configuration file
36     self.cfg_fname = None
37     # Dictionary with configuration parameters
38     self.cfg_params = {}
39    
40     # Current working directory
41     self.cwd = os.getcwd()+'/'
42     # Current time in format 'yymmdd_hhmmss'
43     self.current_time = time.strftime('%y%m%d_%H%M%S',
44     time.localtime(time.time()))
45    
46 nsmirnov 1.3 # Session name (?) Do we need this ?
47 nsmirnov 1.1 self.name = '0'
48    
49     # Job type
50     self.job_type_name = None
51    
52     # Continuation flag
53     self.flag_continue = 0
54    
55     # quiet mode, i.e. no output on screen
56     self.flag_quiet = 0
57     # produce more output
58     self.debug_level = 0
59    
60 nsmirnov 1.3 # Scheduler name, e.g. 'edg', 'lsf'
61 fanzago 1.53 # self.scheduler_name = ''
62 nsmirnov 1.1
63 nsmirnov 1.3 self.initialize_(opts)
64 nsmirnov 1.1
65     return
66    
67 nsmirnov 1.7 def version():
68 nsmirnov 1.1 return common.prog_version_str
69    
70 nsmirnov 1.7 version = staticmethod(version)
71    
72 nsmirnov 1.3 def initialize_(self, opts):
73 nsmirnov 1.1
74     # Process the '-continue' option first because
75     # in the case of continuation the CRAB configuration
76     # parameters are loaded from already existing Working Space.
77 nsmirnov 1.3 self.processContinueOption_(opts)
78 nsmirnov 1.1
79     # Process ini-file first, then command line options
80     # because they override ini-file settings.
81    
82 nsmirnov 1.3 self.processIniFile_(opts)
83 nsmirnov 1.1
84 nsmirnov 1.3 if self.flag_continue: opts = self.loadConfiguration_(opts)
85 nsmirnov 1.1
86 nsmirnov 1.3 self.processOptions_(opts)
87 nsmirnov 1.1
88     if not self.flag_continue:
89 nsmirnov 1.3 self.createWorkingSpace_()
90 slacapra 1.9 optsToBeSaved={}
91     for it in opts.keys():
92     if (it in self.main_actions) or (it in self.aux_actions) or (it == '-debug'):
93     pass
94     else:
95     optsToBeSaved[it]=opts[it]
96     common.work_space.saveConfiguration(optsToBeSaved, self.cfg_fname)
97 nsmirnov 1.1 pass
98    
99     # At this point all configuration options have been read.
100    
101     args = string.join(sys.argv,' ')
102 slacapra 1.11
103 nsmirnov 1.3 self.updateHistory_(args)
104 slacapra 1.11
105 nsmirnov 1.3 self.createLogger_(args)
106 slacapra 1.11
107 nsmirnov 1.1 common.jobDB = JobDB()
108 corvo 1.51
109 corvo 1.63 if int(self.cfg_params['USER.activate_monalisa']): self.cfg_params['apmon'] = ApmonIf()
110    
111 nsmirnov 1.3 if self.flag_continue:
112 slacapra 1.12 try:
113     common.jobDB.load()
114 corvo 1.50 self.cfg_params['taskId'] = common.jobDB._jobs[0].taskId
115 slacapra 1.12 common.logger.debug(6, str(common.jobDB))
116     except DBException,e:
117     pass
118 nsmirnov 1.3 pass
119 slacapra 1.11
120 nsmirnov 1.3 self.createScheduler_()
121 slacapra 1.11
122 nsmirnov 1.3 if common.logger.debugLevel() >= 6:
123     common.logger.debug(6, 'Used properties:')
124     keys = self.cfg_params.keys()
125     keys.sort()
126     for k in keys:
127     if self.cfg_params[k]:
128 corvo 1.42 common.logger.debug(6, ' '+k+' : '+str(self.cfg_params[k]))
129 nsmirnov 1.3 pass
130     else:
131     common.logger.debug(6, ' '+k+' : ')
132     pass
133     pass
134     common.logger.debug(6, 'End of used properties.\n')
135     pass
136     self.initializeActions_(opts)
137 nsmirnov 1.1 return
138    
139 nsmirnov 1.3 def processContinueOption_(self,opts):
140 nsmirnov 1.1
141     continue_dir = None
142 nsmirnov 1.4
143     # Look for the '-continue' option.
144    
145 nsmirnov 1.1 for opt in opts.keys():
146     if ( opt in ('-continue','-c') ):
147     self.flag_continue = 1
148     val = opts[opt]
149     if val:
150     if val[0] == '/': continue_dir = val # abs path
151     else: continue_dir = self.cwd + val # rel path
152     pass
153 nsmirnov 1.4 break
154     pass
155    
156     # Look for actions which has sense only with '-continue'
157    
158     if not self.flag_continue:
159     for opt in opts.keys():
160 slacapra 1.6 if ( opt in (self.aux_actions) ):
161 nsmirnov 1.4 self.flag_continue = 1
162     break
163 nsmirnov 1.1 pass
164     pass
165 slacapra 1.6 submit_flag=0
166     create_flag=0
167     for opt in opts.keys():
168     if opt == "-submit": submit_flag=1
169     if opt == "-create": create_flag=1
170     pass
171     if (submit_flag and not create_flag):
172 nsmirnov 1.7 msg = "'-submit' must be used with either '-create' or '-continue'."
173     raise CrabException(msg)
174 slacapra 1.6 pass
175 nsmirnov 1.1
176     if not self.flag_continue: return
177    
178     if not continue_dir:
179     prefix = common.prog_name + '_' + self.name + '_'
180     continue_dir = findLastWorkDir(prefix)
181     pass
182    
183     if not continue_dir:
184     raise CrabException('Cannot find last working directory.')
185    
186     if not os.path.exists(continue_dir):
187     msg = 'Cannot continue because the working directory <'
188     msg += continue_dir
189     msg += '> does not exist.'
190     raise CrabException(msg)
191    
192     # Instantiate WorkSpace
193 fanzago 1.49 common.work_space = WorkSpace(continue_dir, self.cfg_params)
194 nsmirnov 1.1
195     return
196    
197 nsmirnov 1.3 def processIniFile_(self, opts):
198 nsmirnov 1.1 """
199     Processes a configuration INI-file.
200     """
201    
202     # Extract cfg-file name from the cmd-line options.
203    
204     for opt in opts.keys():
205     if ( opt == '-cfg' ):
206     if self.flag_continue:
207     raise CrabException('-continue and -cfg cannot coexist.')
208     if opts[opt] : self.cfg_fname = opts[opt]
209     else : usage()
210     pass
211    
212     elif ( opt == '-name' ):
213     self.name = opts[opt]
214     pass
215    
216     pass
217    
218     # Set default cfg-fname
219    
220     if self.cfg_fname == None:
221     if self.flag_continue:
222     self.cfg_fname = common.work_space.cfgFileName()
223     else:
224     self.cfg_fname = common.prog_name+'.cfg'
225     pass
226     pass
227    
228     # Load cfg-file
229    
230     if string.lower(self.cfg_fname) != 'none':
231     if os.path.exists(self.cfg_fname):
232     self.cfg_params = loadConfig(self.cfg_fname)
233 corvo 1.64 self.cfg_params['user'] = os.environ['USER']
234 nsmirnov 1.1 pass
235     else:
236     msg = 'cfg-file '+self.cfg_fname+' not found.'
237     raise CrabException(msg)
238     pass
239     pass
240    
241     # process the [CRAB] section
242    
243     lhp = len('CRAB.')
244     for k in self.cfg_params.keys():
245     if len(k) >= lhp and k[:lhp] == 'CRAB.':
246     opt = '-'+k[lhp:]
247     if len(opt) >= 3 and opt[:3] == '-__': continue
248     if opt not in opts.keys():
249     opts[opt] = self.cfg_params[k]
250     pass
251     pass
252     pass
253    
254     return
255    
256 nsmirnov 1.3 def processOptions_(self, opts):
257 nsmirnov 1.1 """
258     Processes the command-line options.
259     """
260    
261     for opt in opts.keys():
262     val = opts[opt]
263    
264 nsmirnov 1.3 # Skip actions, they are processed later in initializeActions_()
265     if opt in self.main_actions:
266     self.cfg_params['CRAB.'+opt[1:]] = val
267     continue
268     if opt in self.aux_actions:
269     self.cfg_params['CRAB.'+opt[1:]] = val
270     continue
271 nsmirnov 1.1
272    
273     elif ( opt == '-cfg' ):
274     pass
275    
276     elif ( opt in ('-continue', '-c') ):
277 nsmirnov 1.4 # Already processed in processContinueOption_()
278 nsmirnov 1.1 pass
279    
280     elif ( opt == '-jobtype' ):
281     if val : self.job_type_name = string.upper(val)
282     else : usage()
283     pass
284    
285     elif ( opt == '-Q' ):
286     self.flag_quiet = 1
287     pass
288    
289     elif ( opt == '-debug' ):
290 slacapra 1.6 if val: self.debug_level = int(val)
291     else: self.debug_level = 1
292 nsmirnov 1.1 pass
293    
294     elif ( opt == '-scheduler' ):
295 fanzago 1.53 if val:
296     self.scheduler_name = 'boss'
297     self.flag_useboss = 1
298 nsmirnov 1.1 else:
299     print common.prog_name+". No value for '-scheduler'."
300     usage()
301     pass
302     pass
303 slacapra 1.22
304 fanzago 1.53 # elif ( opt in ('-use_boss', '-useboss') ):
305     # if ( val == '1' ):
306     # self.scheduler_name = 'boss'
307     # pass
308     # elif ( val == '0' ):
309     # pass
310     # else:
311     # print common.prog_name+'. Bad flag for -use_boss option:',\
312     # val,'Possible values are 0(=No) or 1(=Yes)'
313     # usage()
314     # pass
315     # pass
316 fanzago 1.14
317 nsmirnov 1.3 elif string.find(opt,'.') == -1:
318     print common.prog_name+'. Unrecognized option '+opt
319     usage()
320     pass
321 nsmirnov 1.1
322 nsmirnov 1.3 # Override config parameters from INI-file with cmd-line params
323     if string.find(opt,'.') == -1 :
324     self.cfg_params['CRAB.'+opt[1:]] = val
325 nsmirnov 1.1 pass
326 nsmirnov 1.3 else:
327 nsmirnov 1.1 # Command line parameters in the form -SECTION.ENTRY=VALUE
328     self.cfg_params[opt[1:]] = val
329     pass
330     pass
331     return
332    
333 slacapra 1.8 def parseRange_(self, aRange):
334 nsmirnov 1.4 """
335 slacapra 1.8 Takes as the input a string with a range defined in any of the following
336     way, including combination, and return a tuple with the ints defined
337     according to following table. A consistency check is done.
338     NB: the first job is "1", not "0".
339     'all' -> [1,2,..., NJobs]
340     '' -> [1,2,..., NJobs]
341     'n1' -> [n1]
342     'n1-n2' -> [n1, n1+1, n1+2, ..., n2-1, n2]
343     'n1,n2' -> [n1, n2]
344     'n1,n2-n3' -> [n1, n2, n2+1, n2+2, ..., n3-1, n3]
345     """
346     result = []
347    
348 slacapra 1.9 common.logger.debug(5,"parseRange_ "+str(aRange))
349     if aRange=='all' or aRange==None or aRange=='':
350 slacapra 1.73 result=range(1,common.jobDB.nJobs()+1)
351 slacapra 1.8 return result
352 slacapra 1.9 elif aRange=='0':
353     return result
354 slacapra 1.8
355     subRanges = string.split(aRange, ',')
356     for subRange in subRanges:
357     result = result+self.parseSimpleRange_(subRange)
358    
359     if self.checkUniqueness_(result):
360     return result
361     else:
362 slacapra 1.33 common.logger.message("Error "+result)
363 slacapra 1.8 return []
364    
365     def checkUniqueness_(self, list):
366     """
367 slacapra 1.9 check if a list contains only unique elements
368 slacapra 1.8 """
369    
370     uniqueList = []
371     # use a list comprehension statement (takes a while to understand)
372    
373     [uniqueList.append(it) for it in list if not uniqueList.count(it)]
374    
375     return (len(list)==len(uniqueList))
376    
377     def parseSimpleRange_(self, aRange):
378     """
379     Takes as the input a string with two integers separated by
380     the minus sign and returns the tuple with these numbers:
381     'n1-n2' -> [n1, n1+1, n1+2, ..., n2-1, n2]
382     'n1' -> [n1]
383     """
384     (start, end) = (None, None)
385    
386     result = []
387     minus = string.find(aRange, '-')
388     if ( minus < 0 ):
389     if isInt(aRange) and int(aRange)>0:
390 fanzago 1.37 # FEDE
391     #result.append(int(aRange)-1)
392     ###
393 slacapra 1.62 result.append(int(aRange))
394 slacapra 1.8 else:
395 spiga 1.47 common.logger.message("parseSimpleRange_ ERROR "+aRange)
396     usage()
397     pass
398    
399 nsmirnov 1.4 pass
400     else:
401 slacapra 1.8 (start, end) = string.split(aRange, '-')
402     if isInt(start) and isInt(end) and int(start)>0 and int(start)<int(end):
403 spiga 1.38 #result=range(int(start)-1, int(end))
404     result=range(int(start), int(end)+1) #Daniele
405 slacapra 1.8 else:
406 slacapra 1.33 common.logger.message("parseSimpleRange_ ERROR "+start+end)
407 slacapra 1.8
408     return result
409 nsmirnov 1.4
410 nsmirnov 1.3 def initializeActions_(self, opts):
411 nsmirnov 1.1 """
412     For each user action instantiate a corresponding
413     object and put it in the action dictionary.
414     """
415 fanzago 1.53 # for opt in opts.keys():
416     # self.flag_useboss = 0
417     # if ( opt == '-use_boss'):
418     # val = opts[opt]
419     # if ( val == '1' ):
420     # self.flag_useboss = 1
421     # common.logger.message('Using BOSS')
422     # pass
423     # else:
424     # self.flag_useboss = 0
425     # pass
426 spiga 1.15
427     for opt in opts.keys():
428    
429 nsmirnov 1.1 val = opts[opt]
430 spiga 1.15
431    
432     if ( opt == '-create' ):
433 slacapra 1.22 ncjobs = 0
434 nsmirnov 1.1 if val:
435     if ( isInt(val) ):
436     ncjobs = int(val)
437     elif ( val == 'all'):
438     ncjobs = val
439     else:
440 nsmirnov 1.5 msg = 'Bad creation bunch size <'+str(val)+'>\n'
441     msg += ' Must be an integer or "all"'
442 slacapra 1.8 msg += ' Generic range is not allowed"'
443 nsmirnov 1.5 raise CrabException(msg)
444 nsmirnov 1.1 pass
445     else: ncjobs = 'all'
446    
447     if ncjobs != 0:
448 nsmirnov 1.3 # Instantiate Creator object
449 fanzago 1.56 self.creator = Creator(self.job_type_name,
450 gutsche 1.61 self.cfg_params,
451     ncjobs)
452 fanzago 1.56 self.actions[opt] = self.creator
453 nsmirnov 1.3
454     # Initialize the JobDB object if needed
455 nsmirnov 1.1 if not self.flag_continue:
456 fanzago 1.56 common.jobDB.create(self.creator.nJobs())
457 nsmirnov 1.1 pass
458 nsmirnov 1.3
459     # Create and initialize JobList
460    
461     common.job_list = JobList(common.jobDB.nJobs(),
462 fanzago 1.56 self.creator.jobType())
463 nsmirnov 1.3
464     common.job_list.setScriptNames(self.job_type_name+'.sh')
465     common.job_list.setJDLNames(self.job_type_name+'.jdl')
466 gutsche 1.66 common.job_list.setCfgNames(self.creator.jobType().configFilename())
467 slacapra 1.9
468 fanzago 1.56 self.creator.writeJobsSpecsToDB()
469 nsmirnov 1.1 pass
470 nsmirnov 1.3 pass
471 nsmirnov 1.1
472     elif ( opt == '-submit' ):
473 slacapra 1.23
474     # total jobs
475     # get the first not already submitted
476 slacapra 1.24 common.logger.debug(5,'Total jobs '+str(common.jobDB.nJobs()))
477     lastSubmittedJob=0
478     for nj in range(common.jobDB.nJobs()):
479 spiga 1.71 if (common.jobDB.status(nj) in ['S','K','Y','A','D']):
480 slacapra 1.24 lastSubmittedJob +=1
481 slacapra 1.30 else: break
482 slacapra 1.24 # count job from 1
483 slacapra 1.30 totalJobsSubmittable = common.jobDB.nJobs()-lastSubmittedJob
484 slacapra 1.24 common.logger.debug(5,'lastSubmittedJob '+str(lastSubmittedJob))
485     common.logger.debug(5,'totalJobsSubmittable '+str(totalJobsSubmittable))
486 slacapra 1.23
487 slacapra 1.24 nsjobs = lastSubmittedJob+totalJobsSubmittable
488 slacapra 1.23 # get user request
489 slacapra 1.22 if val:
490     if ( isInt(val) ):
491 slacapra 1.24 tmp = int(val)
492 slacapra 1.72 if (tmp > totalJobsSubmittable):
493 slacapra 1.24 common.logger.message('asking to submit '+str(tmp)+' jobs, but only '+str(totalJobsSubmittable)+' left: submitting those')
494     pass
495     else:
496     nsjobs=lastSubmittedJob+int(val)
497 slacapra 1.23 elif (val=='all'):
498     pass
499 slacapra 1.22 else:
500     msg = 'Bad submission option <'+str(val)+'>\n'
501     msg += ' Must be an integer or "all"'
502     msg += ' Generic range is not allowed"'
503     raise CrabException(msg)
504     pass
505 slacapra 1.24 common.logger.debug(5,'nsjobs '+str(nsjobs))
506 slacapra 1.23
507     # submit N from last submitted job
508 slacapra 1.24 nj_list = range(lastSubmittedJob, nsjobs)
509     common.logger.debug(5,'nj_list '+str(nj_list))
510 nsmirnov 1.5
511     if len(nj_list) != 0:
512 nsmirnov 1.3 # Instantiate Submitter object
513 gutsche 1.61 # self.actions[opt] = Submitter(self.cfg_params, nj_list, self.job_type)
514 corvo 1.57 self.actions[opt] = Submitter(self.cfg_params, nj_list)
515 nsmirnov 1.3 # Create and initialize JobList
516     if len(common.job_list) == 0 :
517     common.job_list = JobList(common.jobDB.nJobs(),
518     None)
519     common.job_list.setJDLNames(self.job_type_name+'.jdl')
520     pass
521 nsmirnov 1.1 pass
522 slacapra 1.30 pass
523 nsmirnov 1.1
524 nsmirnov 1.4 elif ( opt == '-list' ):
525 slacapra 1.8 jobs = self.parseRange_(val)
526 slacapra 1.73 print jobs
527 slacapra 1.8
528     common.jobDB.dump(jobs)
529 nsmirnov 1.4 pass
530    
531 slacapra 1.67 elif ( opt == '-printId' ):
532     jobs = self.parseRange_(val)
533    
534     for nj in jobs:
535     #nj parte da 1 --> nj = internal_id di boss
536 slacapra 1.73 st = common.jobDB.status(nj-1)
537 slacapra 1.67 if st == 'S' or st == 'A':
538 slacapra 1.73 id = common.scheduler.boss_SID(nj)
539     print "Job: ",nj," Id = ", id
540 slacapra 1.67 else:
541 slacapra 1.73 print "Job: ",nj," No ID yet"
542 slacapra 1.67 pass
543    
544 nsmirnov 1.4 elif ( opt == '-status' ):
545 slacapra 1.8 jobs = self.parseRange_(val)
546    
547 slacapra 1.9 if len(jobs) != 0:
548 spiga 1.16 if ( self.flag_useboss == 1 ):
549 corvo 1.42 self.actions[opt] = StatusBoss(self.cfg_params)
550 spiga 1.15 else:
551 corvo 1.46 self.actions[opt] = Status(self.cfg_params, jobs)
552 spiga 1.15 pass
553 nsmirnov 1.1 pass
554     pass
555 slacapra 1.22
556 nsmirnov 1.4 elif ( opt == '-kill' ):
557 slacapra 1.8
558 spiga 1.31 if ( self.flag_useboss == 1 ):
559     if val:
560     if val =='all':
561 fanzago 1.37 allBoss_id = common.scheduler.listBoss()
562     jobs = allBoss_id.keys()
563 spiga 1.31 else:
564     jobs = self.parseRange_(val)
565     common.scheduler.cancel(jobs)
566     else:
567     common.logger.message("Warning: with '-kill' you _MUST_ specify a job range or 'all'")
568     else:
569     if val:
570     jobs = self.parseRange_(val)
571    
572     for nj in jobs:
573     st = common.jobDB.status(nj)
574 gutsche 1.61 if st == 'S' or st == 'A':
575 spiga 1.31 jid = common.jobDB.jobId(nj)
576     common.logger.message("Killing job # "+`(nj+1)`)
577     common.scheduler.cancel(jid)
578     common.jobDB.setStatus(nj, 'K')
579     pass
580 slacapra 1.9 pass
581 spiga 1.31
582     common.jobDB.save()
583 nsmirnov 1.4 pass
584 spiga 1.31 else:
585     common.logger.message("Warning: with '-kill' you _MUST_ specify a job range or 'all'")
586 nsmirnov 1.1
587 slacapra 1.70 elif ( opt == '-getoutput' or opt == '-get'):
588 slacapra 1.8
589 spiga 1.20 if ( self.flag_useboss == 1 ):
590     if val=='all' or val==None or val=='':
591 fanzago 1.37 allBoss_id = common.scheduler.listBoss()
592     jobs = allBoss_id.keys()
593 spiga 1.20 else:
594     jobs = self.parseRange_(val)
595     common.scheduler.getOutput(jobs)
596     else:
597     jobs = self.parseRange_(val)
598 spiga 1.13
599 slacapra 1.22 ## also this: create a ActorClass (GetOutput)
600 spiga 1.20 jobs_done = []
601     for nj in jobs:
602     st = common.jobDB.status(nj)
603     if st == 'D':
604 slacapra 1.12 jobs_done.append(nj)
605 spiga 1.20 pass
606     elif st == 'S':
607     jid = common.jobDB.jobId(nj)
608     currStatus = common.scheduler.queryStatus(jid)
609     if currStatus=="Done":
610     jobs_done.append(nj)
611     else:
612     msg = 'Job # '+`(nj+1)`+' submitted but still status '+currStatus+' not possible to get output'
613     common.logger.message(msg)
614     pass
615 slacapra 1.9 else:
616 spiga 1.20 # common.logger.message('Jobs #'+`(nj+1)`+' has status '+st+' not possible to get output')
617     pass
618 slacapra 1.12 pass
619    
620 spiga 1.20 for nj in jobs_done:
621 spiga 1.15 jid = common.jobDB.jobId(nj)
622     dir = common.scheduler.getOutput(jid)
623     common.jobDB.setStatus(nj, 'Y')
624 slacapra 1.12
625     # Rename the directory with results to smth readable
626     new_dir = common.work_space.resDir()
627 gutsche 1.61 if ( dir != '' ) :
628     try:
629     files = os.listdir(dir)
630     for file in files:
631     os.rename(dir+'/'+file, new_dir+'/'+file)
632     os.rmdir(dir)
633     except OSError, e:
634     msg = 'rename files from '+dir+' to '+new_dir+' error: '
635     msg += str(e)
636     common.logger.message(msg)
637     # ignore error
638     pass
639 slacapra 1.12 pass
640 slacapra 1.22 ###
641    
642 spiga 1.13 resFlag = 0
643     exCode = common.scheduler.getExitStatus(jid)
644 spiga 1.34 Statistic.Monitor('retrieved',resFlag,jid,exCode)
645 slacapra 1.12
646     msg = 'Results of Job # '+`(nj+1)`+' are in '+new_dir
647     common.logger.message(msg)
648 nsmirnov 1.4 pass
649    
650     common.jobDB.save()
651     pass
652    
653     elif ( opt == '-resubmit' ):
654 fanzago 1.37 if ( self.flag_useboss == 1 ):
655     if val=='all' or val==None or val=='':
656     allBoss_id = common.scheduler.listBoss()
657     jobs = allBoss_id.keys()
658     else:
659     jobs = self.parseRange_(val)
660     else:
661     if val:
662     jobs = self.parseRange_(val)
663    
664 slacapra 1.11 if val:
665     # create a list of jobs to be resubmitted.
666 slacapra 1.8
667 slacapra 1.22 ### as before, create a Resubmittter Class
668 spiga 1.60 allBoss_id = common.scheduler.listBoss()
669     maxIndex = allBoss_id.keys()
670 slacapra 1.11 nj_list = []
671     for nj in jobs:
672 spiga 1.60 if int(nj) <= int(len(maxIndex)) :
673     st = common.jobDB.status(int(nj)-1)
674     if st in ['K','A']:
675     nj_list.append(int(nj)-1)
676     common.jobDB.setStatus(int(nj)-1,'C')
677     elif st == 'Y':
678     common.scheduler.moveOutput(nj)
679     nj_list.append(int(nj)-1)
680     st = common.jobDB.setStatus(int(nj)-1,'RC')
681     elif st in ['C','X']:
682 gutsche 1.61 common.logger.message('Job #'+`int(nj)`+' has status '+crabJobStatusToString(st)+' not yet submitted!!!')
683 spiga 1.60 pass
684     elif st == 'D':
685     common.logger.message('Job #'+`int(nj)`+' has status '+crabJobStatusToString(st)+' must be retrieved before resubmission')
686     else:
687     common.logger.message('Job #'+`nj`+' has status '+crabJobStatusToString(st)+' must be "killed" before resubmission')
688 slacapra 1.11 else:
689 gutsche 1.61 common.logger.message('Job #'+`int(nj)`+' no possible to resubmit!! out of range')
690 fanzago 1.37 if len(common.job_list) == 0 :
691     common.job_list = JobList(common.jobDB.nJobs(),None)
692     common.job_list.setJDLNames(self.job_type_name+'.jdl')
693     pass
694    
695 slacapra 1.11 if len(nj_list) != 0:
696 fanzago 1.37 common.scheduler.resubmit(nj_list)
697 slacapra 1.11 # Instantiate Submitter object
698 fanzago 1.58 self.actions[opt] = Submitter(self.cfg_params, nj_list)
699 slacapra 1.11
700 slacapra 1.8 pass
701     pass
702 slacapra 1.11 else:
703     common.logger.message("Warning: with '-resubmit' you _MUST_ specify a job range or 'all'")
704 spiga 1.60 common.logger.message("WARNING: _all_ job specified in the range will be resubmitted!!!")
705 slacapra 1.11 pass
706 fanzago 1.37 common.jobDB.save()
707 slacapra 1.8 pass
708 gutsche 1.61
709 slacapra 1.8 elif ( opt == '-cancelAndResubmit' ):
710 nsmirnov 1.5
711 spiga 1.47 if ( self.flag_useboss == 1 ):
712     if val:
713     if val =='all':
714     allBoss_id = common.scheduler.listBoss()
715     jobs = allBoss_id.keys()
716     else:
717     jobs = self.parseRange_(val)
718     # kill submitted jobs
719     common.scheduler.cancel(jobs)
720     else:
721     common.logger.message("Warning: with '-cancelAndResubmit' you _MUST_ specify a job range or 'all'")
722     else:
723     if val:
724     jobs = self.parseRange_(val)
725     else:
726     common.logger.message("Warning: with '-kill' you _MUST_ specify a job range or 'all'")
727 nsmirnov 1.5 pass
728    
729 spiga 1.47 # resubmit cancelled jobs.
730     if val:
731     nj_list = []
732     for nj in jobs:
733     if ( self.flag_useboss != 1 ):
734     st = common.jobDB.status(nj)
735     if st == 'S':
736     jid = common.jobDB.jobId(nj)
737     common.scheduler.cancel(jid)
738     st = 'K'
739     common.jobDB.setStatus(nj, st)
740     pass
741     common.jobDB.save()
742     pass
743     st = common.jobDB.status(int(nj)-1)
744     if st in ['K','A']:
745     nj_list.append(int(nj)-1)
746     common.jobDB.setStatus(int(nj)-1,'C')
747     elif st == 'Y':
748     common.scheduler.moveOutput(nj)
749     nj_list.append(int(nj)-1)
750     st = common.jobDB.setStatus(int(nj)-1,'RC')
751     elif st in ['C','X']:
752     common.logger.message('Job #'+`int(nj)`+' has status '+crabJobStatusToString(st)+' not yet submitted!!!')
753     pass
754     elif st == 'D':
755     common.logger.message('Job #'+`int(nj)`+' has status '+crabJobStatusToString(st)+' must be retrieved before resubmission')
756     else:
757     common.logger.message('Job #'+`nj`+' has status '+crabJobStatusToString(st)+' must be "killed" before resubmission')
758     pass
759    
760 nsmirnov 1.5 if len(common.job_list) == 0 :
761 spiga 1.47 common.job_list = JobList(common.jobDB.nJobs(),None)
762 nsmirnov 1.5 common.job_list.setJDLNames(self.job_type_name+'.jdl')
763     pass
764 spiga 1.47
765     if len(nj_list) != 0:
766     common.scheduler.resubmit(nj_list)
767 fanzago 1.58 self.actions[opt] = Submitter(self.cfg_params, nj_list)
768 spiga 1.47 pass
769     pass
770     else:
771     common.logger.message("WARNING: _all_ job specified in the rage will be cancelled and resubmitted!!!")
772 nsmirnov 1.5 pass
773 spiga 1.47 common.jobDB.save()
774 nsmirnov 1.4 pass
775 spiga 1.47
776 slacapra 1.28 elif ( opt == '-testJdl' ):
777 slacapra 1.8 jobs = self.parseRange_(val)
778     nj_list = []
779     for nj in jobs:
780 slacapra 1.62 st = common.jobDB.status(nj-1)
781     if st == 'C': nj_list.append(nj-1)
782 slacapra 1.8 pass
783    
784     if len(nj_list) != 0:
785     # Instantiate Submitter object
786     self.actions[opt] = Checker(self.cfg_params, nj_list)
787    
788     # Create and initialize JobList
789    
790     if len(common.job_list) == 0 :
791     common.job_list = JobList(common.jobDB.nJobs(), None)
792     common.job_list.setJDLNames(self.job_type_name+'.jdl')
793     pass
794     pass
795    
796 slacapra 1.9 elif ( opt == '-postMortem' ):
797     jobs = self.parseRange_(val)
798     nj_list = []
799     for nj in jobs:
800 fanzago 1.45 # fede: nj scala di uno perche' e' l'internal id di boss
801     # ed il jobDB parte da zero ...
802     st = common.jobDB.status(int(nj)-1)
803     if st not in ['X', 'C']: nj_list.append(int(nj))
804 slacapra 1.9 pass
805    
806     if len(nj_list) != 0:
807     # Instantiate Submitter object
808 spiga 1.35 self.actions[opt] = PostMortem(self.cfg_params, nj_list,self.flag_useboss)
809 slacapra 1.9
810     # Create and initialize JobList
811    
812     if len(common.job_list) == 0 :
813     common.job_list = JobList(common.jobDB.nJobs(), None)
814     common.job_list.setJDLNames(self.job_type_name+'.jdl')
815     pass
816     pass
817 slacapra 1.33 else:
818     common.logger.message("No jobs to analyze")
819 slacapra 1.9
820     elif ( opt == '-clean' ):
821     if val != None:
822     raise CrabException("No range allowed for '-clean'")
823    
824 corvo 1.46 theCleaner = Cleaner(self.scheduler_name == 'boss', self.cfg_params)
825 slacapra 1.36 theCleaner.clean()
826 slacapra 1.9
827 nsmirnov 1.1 pass
828     return
829    
830 nsmirnov 1.3 def createWorkingSpace_(self):
831 slacapra 1.9 new_dir = ''
832    
833     try:
834     new_dir = self.cfg_params['USER.ui_working_dir']
835 corvo 1.50 self.cfg_params['taskId'] = self.cfg_params['user'] + '_' + string.split(new_dir, '/')[len(string.split(new_dir, '/')) - 1] + '_' + self.current_time
836 fanzago 1.48 if os.path.exists(new_dir):
837     if os.listdir(new_dir):
838     msg = new_dir + ' already exists and is not empty. Please remove it before create new task'
839     raise CrabException(msg)
840 slacapra 1.9 except KeyError:
841     new_dir = common.prog_name + '_' + self.name + '_' + self.current_time
842 corvo 1.50 self.cfg_params['taskId'] = self.cfg_params['user'] + '_' + new_dir
843 slacapra 1.9 new_dir = self.cwd + new_dir
844     pass
845 fanzago 1.49 common.work_space = WorkSpace(new_dir, self.cfg_params)
846 nsmirnov 1.1 common.work_space.create()
847     return
848    
849 nsmirnov 1.3 def loadConfiguration_(self, opts):
850 nsmirnov 1.1
851     save_opts = common.work_space.loadSavedOptions()
852    
853     # Override saved options with new command-line options
854    
855     for k in opts.keys():
856     save_opts[k] = opts[k]
857     pass
858    
859     # Return updated options
860     return save_opts
861    
862 nsmirnov 1.3 def createLogger_(self, args):
863 nsmirnov 1.1
864     log = Logger()
865     log.quiet(self.flag_quiet)
866     log.setDebugLevel(self.debug_level)
867     log.write(args+'\n')
868 nsmirnov 1.3 log.message(self.headerString_())
869 nsmirnov 1.1 log.flush()
870     common.logger = log
871     return
872    
873 nsmirnov 1.3 def updateHistory_(self, args):
874 nsmirnov 1.1 history_fname = common.prog_name+'.history'
875     history_file = open(history_fname, 'a')
876     history_file.write(self.current_time+': '+args+'\n')
877     history_file.close()
878     return
879    
880 nsmirnov 1.3 def headerString_(self):
881 nsmirnov 1.1 """
882     Creates a string describing program options either given in
883     the command line or their default values.
884     """
885     header = common.prog_name + ' (version ' + common.prog_version_str + \
886     ') running on ' + \
887     time.ctime(time.time())+'\n\n' + \
888     common.prog_name+'. Working options:\n'
889 fanzago 1.59 #print self.job_type_name
890 nsmirnov 1.1 header = header +\
891     ' scheduler ' + self.scheduler_name + '\n'+\
892     ' job type ' + self.job_type_name + '\n'+\
893     ' working directory ' + common.work_space.topDir()\
894     + '\n'
895     return header
896    
897 nsmirnov 1.3 def createScheduler_(self):
898 nsmirnov 1.1 """
899     Creates a scheduler object instantiated by its name.
900     """
901     klass_name = 'Scheduler' + string.capitalize(self.scheduler_name)
902     file_name = klass_name
903     try:
904     klass = importName(file_name, klass_name)
905     except KeyError:
906     msg = 'No `class '+klass_name+'` found in file `'+file_name+'.py`'
907     raise CrabException(msg)
908     except ImportError, e:
909     msg = 'Cannot create scheduler '+self.scheduler_name
910     msg += ' (file: '+file_name+', class '+klass_name+'):\n'
911     msg += str(e)
912     raise CrabException(msg)
913    
914     common.scheduler = klass()
915     common.scheduler.configure(self.cfg_params)
916     return
917    
918 corvo 1.55 # def createJobtype_(self):
919     # """
920     # Create the jobtype specified in the crab.cfg file
921     # """
922     # file_name = 'cms_'+ string.lower(self.job_type_name)
923     # klass_name = string.capitalize(self.job_type_name)
924 gutsche 1.61 #
925 corvo 1.55 # try:
926     # klass = importName(file_name, klass_name)
927     # except KeyError:
928     # msg = 'No `class '+klass_name+'` found in file `'+file_name+'.py`'
929     # raise CrabException(msg)
930     # except ImportError, e:
931     # msg = 'Cannot create job type '+self.job_type_name
932     # msg += ' (file: '+file_name+', class '+klass_name+'):\n'
933     # msg += str(e)
934     # raise CrabException(msg)
935     # job_type = klass(self.cfg_params)
936     # return job_type
937 corvo 1.51
938 nsmirnov 1.1 def run(self):
939     """
940     For each
941     """
942    
943     for act in self.main_actions:
944     if act in self.actions.keys(): self.actions[act].run()
945     pass
946    
947     for act in self.aux_actions:
948     if act in self.actions.keys(): self.actions[act].run()
949     pass
950     return
951    
952     ###########################################################################
953     def processHelpOptions(opts):
954    
955 slacapra 1.11 if len(opts):
956     for opt in opts.keys():
957     if opt in ('-v', '-version', '--version') :
958     print Crab.version()
959     return 1
960     if opt in ('-h','-help','--help') :
961     if opts[opt] : help(opts[opt])
962     else: help()
963     return 1
964     else:
965     usage()
966 nsmirnov 1.1
967     return 0
968    
969     ###########################################################################
970     if __name__ == '__main__':
971    
972 fanzago 1.18
973 fanzago 1.19 # # Initial settings for Python modules. Avoid appending manually lib paths.
974     # try:
975     # path=os.environ['EDG_WL_LOCATION']
976     # except:
977     # print "Error: Please set the EDG_WL_LOCATION environment variable pointing to the userinterface installation path"
978     # sys.exit(1)
979     #
980     # libPath=os.path.join(path, "lib")
981     # sys.path.append(libPath)
982     # libPath=os.path.join(path, "lib", "python")
983     # sys.path.append(libPath)
984 fanzago 1.18
985    
986 nsmirnov 1.1 # Parse command-line options and create a dictionary with
987     # key-value pairs.
988    
989     options = parseOptions(sys.argv[1:])
990    
991     # Process "help" options, such as '-help', '-version'
992    
993 slacapra 1.11 if processHelpOptions(options) : sys.exit(0)
994 nsmirnov 1.1
995     # Create, initialize, and run a Crab object
996    
997     try:
998 nsmirnov 1.3 crab = Crab(options)
999 nsmirnov 1.1 crab.run()
1000 corvo 1.54 crab.cfg_params['apmon'].free()
1001 nsmirnov 1.1 except CrabException, e:
1002     print '\n' + common.prog_name + ': ' + str(e) + '\n'
1003     if common.logger:
1004     common.logger.write('ERROR: '+str(e)+'\n')
1005     pass
1006     pass
1007    
1008     pass