CRAB/python/Creator.py

from Actor import Actor
from WorkSpace import WorkSpace
from JobList import JobList
from ScriptWriter import ScriptWriter
from Scheduler import Scheduler
from crab_logger import Logger
from crab_exceptions import *
from crab_util import *
import common

import os, string, math
import time

class Creator(Actor):
    def __init__(self, job_type_name, cfg_params, ncjobs):
        self.job_type_name = job_type_name
        self.job_type = None
        self.cfg_params = cfg_params
        self.total_njobs = 0
        self.total_number_of_events = 0
        self.job_number_of_events = 0
        self.first_event = 0
        self.jobParamsList=[]
     
        self.createJobTypeObject(ncjobs)
        common.logger.debug(5, __name__+": JobType "+self.job_type.name()+" created")

        self.job_type.prepareSteeringCards()
        common.logger.debug(5, __name__+": Steering cards prepared")

        self.total_njobs = self.job_type.numberOfJobs();
        common.logger.debug(5, __name__+": total # of jobs = "+`self.total_njobs`)

        self.ncjobs = ncjobs
        if ncjobs == 'all' : self.ncjobs = self.total_njobs
        if ncjobs > self.total_njobs : self.ncjobs = self.total_njobs
        
        self.job_type_name = self.job_type.name()
 
        common.logger.debug(5, "Creator constructor finished")
        return

    def writeJobsSpecsToDB(self):
        """
        Write firstEvent and maxEvents in the DB for future use
        """

        self.job_type.split(self.jobParamsList)
        return

    def nJobs(self):
        return self.total_njobs
    

    def nJobsL(self):
        jobsL=[]
        for i in range(self.total_njobs):
            jobsL.append(i+1)   
        return jobsL

    def createJobTypeObject(self,ncjobs):
        file_name = 'cms_'+ string.lower(self.job_type_name)
        klass_name = string.capitalize(self.job_type_name)

        try:
            klass = importName(file_name, klass_name)
        except KeyError:
            msg = 'No `class '+klass_name+'` found in file `'+file_name+'.py`'
            raise CrabException(msg)
        except ImportError, e:
            msg = 'Cannot create job type '+self.job_type_name
            msg += ' (file: '+file_name+', class '+klass_name+'):\n'
            msg += str(e)
            raise CrabException(msg)

        self.job_type = klass(self.cfg_params,ncjobs)
        return

    def jobType(self):
        return self.job_type

    def run(self):
        """
        The main method of the class.
        """

        common.logger.debug(5, "Creator::run() called")
        start = time.time()
        # Instantiate ScriptWriter
        script_writer = None
        if self.cfg_params['CRAB.scheduler'].find("glit") != -1: ## checking scheduler: if glite(coll) output_sandbox will be limited
            script_writer = ScriptWriter('crab_template.sh', 1) ## flag that indicates if limit or not
        else:
            script_writer = ScriptWriter('crab_template.sh', 0)

        # Loop over jobs
        argsList = []
        njc = 0
        listID=[]
        listField=[]
        listRunField=[]
        run_jobToSave = {'status' :'C'}
        for nj in range(self.total_njobs):
            output=[]
            if njc == self.ncjobs : break

            common.logger.debug(1,"Creating job # "+`(nj+1)`)
            listRunField.append(run_jobToSave)

            # Prepare configuration file

            self.job_type.modifySteeringCards(nj)
           # outputSandbox=[]
          #  outputSandbox=self.job_type.outputSandbox(nj)
          #  outputSandbox.append('out_files_'+str(nj+1)+'.tgz')
            output.append('out_files_'+str(nj+1)+'.tgz')
           # job_ToSave={'outputFiles': outputSandbox}
            job_ToSave={'outputFiles': output}
            listField.append(job_ToSave)             

            listID.append(nj+1)
            njc = njc + 1
            pass
 
       # ## Not clear why here.. DS
       # self.job_type.setArgsList()
        common._db.updateRunJob_(listID , listRunField ) ## New BL--DS
        common._db.updateJob_(listID, listField ) ## Nes BL--DS


        # Create script (sh)
        script_writer.modifyTemplateScript()
        os.chmod(common._db.queryTask('scriptName'), 0744) ## Modified BL--DS
       # common: write input sandbox --- This is now a task attribute... not per job ## BL--DS

        # SL This should be a list, rather than a string!
        concString = ','
        inSand=''
        if len(self.job_type.inputSandbox(1)): 
            inSand +=   concString.join(self.job_type.inputSandbox(1)) 
        # Sandbox, Start Dir , outputDir
        param = {'globalSandbox': inSand , 'startDirectory': common.work_space.cwdDir() , 'outputDirectory': common.work_space.resDir() }
        common._db.updateTask_(param) 

        # set start Directory


        ####
        common.scheduler.declare(self.total_njobs ) 
        common.scheduler.sched_fix_parameter()
       # common.scheduler.sched_parameter()
       
        stop = time.time()
        common.logger.message('Creating '+str(self.total_njobs)+' jobs, please wait...')

        stop = time.time()
        common.logger.debug(2, "Creation Time: "+str(stop - start))
        common.logger.write("Creation Time: "+str(stop - start))

        msg = '\nTotal of %d jobs created'%njc
        if njc != self.ncjobs: msg = msg + ' from %d requested'%self.ncjobs
        msg = msg + '.\n'
        common.logger.message(msg)
        
        return
Revision:	1.66
Committed:	Sun Apr 20 09:34:40 2008 UTC (17 years ago) by spiga
Content type:	text/x-python
Branch:	MAIN
Changes since 1.65:	+0 -30 lines
Log Message:	reimplemented dashboard communication sending again both pre and post submssion infos, plus many Submitter code reorganization
#	User	Rev	Content
1	nsmirnov	1.1	from Actor import Actor
2			from WorkSpace import WorkSpace
3			from JobList import JobList
4			from ScriptWriter import ScriptWriter
5			from Scheduler import Scheduler
6			from crab_logger import Logger
7			from crab_exceptions import *
8			from crab_util import *
9			import common
10
11	slacapra	1.8	import os, string, math
12	slacapra	1.45	import time
13
14	nsmirnov	1.1	class Creator(Actor):
15	corvo	1.30	def __init__(self, job_type_name, cfg_params, ncjobs):
16	nsmirnov	1.1	self.job_type_name = job_type_name
17	corvo	1.30	self.job_type = None
18	nsmirnov	1.1	self.cfg_params = cfg_params
19	nsmirnov	1.2	self.total_njobs = 0
20			self.total_number_of_events = 0
21			self.job_number_of_events = 0
22	slacapra	1.7	self.first_event = 0
23	gutsche	1.38	self.jobParamsList=[]
24	mcinquil	1.52
25	gutsche	1.42	self.createJobTypeObject(ncjobs)
26	nsmirnov	1.3	common.logger.debug(5, __name__+": JobType "+self.job_type.name()+" created")
27	nsmirnov	1.1
28			self.job_type.prepareSteeringCards()
29	nsmirnov	1.3	common.logger.debug(5, __name__+": Steering cards prepared")
30	nsmirnov	1.1
31	gutsche	1.38	self.total_njobs = self.job_type.numberOfJobs();
32	nsmirnov	1.3	common.logger.debug(5, __name__+": total # of jobs = "+`self.total_njobs`)
33	nsmirnov	1.1
34			self.ncjobs = ncjobs
35			if ncjobs == 'all' : self.ncjobs = self.total_njobs
36			if ncjobs > self.total_njobs : self.ncjobs = self.total_njobs
37	spiga	1.11
38	spiga	1.60	self.job_type_name = self.job_type.name()
39
40	nsmirnov	1.3	common.logger.debug(5, "Creator constructor finished")
41	nsmirnov	1.1	return
42
43	slacapra	1.7	def writeJobsSpecsToDB(self):
44			"""
45			Write firstEvent and maxEvents in the DB for future use
46			"""
47
48	gutsche	1.38	self.job_type.split(self.jobParamsList)
49	nsmirnov	1.2	return
50
51	nsmirnov	1.1	def nJobs(self):
52			return self.total_njobs
53
54	spiga	1.61
55			def nJobsL(self):
56			jobsL=[]
57			for i in range(self.total_njobs):
58			jobsL.append(i+1)
59			return jobsL
60
61	gutsche	1.42	def createJobTypeObject(self,ncjobs):
62	nsmirnov	1.1	file_name = 'cms_'+ string.lower(self.job_type_name)
63			klass_name = string.capitalize(self.job_type_name)
64
65			try:
66			klass = importName(file_name, klass_name)
67			except KeyError:
68			msg = 'No `class '+klass_name+'` found in file `'+file_name+'.py`'
69			raise CrabException(msg)
70			except ImportError, e:
71			msg = 'Cannot create job type '+self.job_type_name
72			msg += ' (file: '+file_name+', class '+klass_name+'):\n'
73			msg += str(e)
74			raise CrabException(msg)
75
76	gutsche	1.42	self.job_type = klass(self.cfg_params,ncjobs)
77	nsmirnov	1.1	return
78	nsmirnov	1.5
79			def jobType(self):
80			return self.job_type
81	nsmirnov	1.1
82			def run(self):
83	nsmirnov	1.4	"""
84			The main method of the class.
85			"""
86	nsmirnov	1.1
87	nsmirnov	1.3	common.logger.debug(5, "Creator::run() called")
88	slacapra	1.45	start = time.time()
89	nsmirnov	1.1	# Instantiate ScriptWriter
90	mcinquil	1.51	script_writer = None
91			if self.cfg_params['CRAB.scheduler'].find("glit") != -1: ## checking scheduler: if glite(coll) output_sandbox will be limited
92			script_writer = ScriptWriter('crab_template.sh', 1) ## flag that indicates if limit or not
93			else:
94			script_writer = ScriptWriter('crab_template.sh', 0)
95	nsmirnov	1.1
96			# Loop over jobs
97	slacapra	1.44	argsList = []
98	nsmirnov	1.1	njc = 0
99	spiga	1.60	listID=[]
100			listField=[]
101	spiga	1.63	listRunField=[]
102			run_jobToSave = {'status' :'C'}
103	nsmirnov	1.4	for nj in range(self.total_njobs):
104	spiga	1.64	output=[]
105	nsmirnov	1.1	if njc == self.ncjobs : break
106
107	slacapra	1.44	common.logger.debug(1,"Creating job # "+`(nj+1)`)
108	spiga	1.63	listRunField.append(run_jobToSave)
109	nsmirnov	1.3
110	nsmirnov	1.1	# Prepare configuration file
111
112	nsmirnov	1.4	self.job_type.modifySteeringCards(nj)
113	spiga	1.64	# outputSandbox=[]
114			# outputSandbox=self.job_type.outputSandbox(nj)
115			# outputSandbox.append('out_files_'+str(nj+1)+'.tgz')
116			output.append('out_files_'+str(nj+1)+'.tgz')
117			# job_ToSave={'outputFiles': outputSandbox}
118			job_ToSave={'outputFiles': output}
119	spiga	1.60	listField.append(job_ToSave)
120	spiga	1.57
121	spiga	1.62	listID.append(nj+1)
122	nsmirnov	1.1	njc = njc + 1
123			pass
124	spiga	1.60
125			# ## Not clear why here.. DS
126			# self.job_type.setArgsList()
127	spiga	1.63	common._db.updateRunJob_(listID , listRunField ) ## New BL--DS
128	spiga	1.60	common._db.updateJob_(listID, listField ) ## Nes BL--DS
129	spiga	1.57
130	slacapra	1.65
131	spiga	1.57	# Create script (sh)
132			script_writer.modifyTemplateScript()
133			os.chmod(common._db.queryTask('scriptName'), 0744) ## Modified BL--DS
134			# common: write input sandbox --- This is now a task attribute... not per job ## BL--DS
135	spiga	1.60
136	slacapra	1.65	# SL This should be a list, rather than a string!
137	spiga	1.60	concString = ','
138			inSand=''
139			if len(self.job_type.inputSandbox(1)):
140			inSand += concString.join(self.job_type.inputSandbox(1))
141	slacapra	1.65	# Sandbox, Start Dir , outputDir
142			param = {'globalSandbox': inSand , 'startDirectory': common.work_space.cwdDir() , 'outputDirectory': common.work_space.resDir() }
143			common._db.updateTask_(param)
144
145			# set start Directory
146	spiga	1.60
147
148	nsmirnov	1.1	####
149	spiga	1.57	common.scheduler.declare(self.total_njobs )
150	spiga	1.60	common.scheduler.sched_fix_parameter()
151			# common.scheduler.sched_parameter()
152
153			stop = time.time()
154	slacapra	1.44	common.logger.message('Creating '+str(self.total_njobs)+' jobs, please wait...')
155	fanzago	1.12
156	slacapra	1.45	stop = time.time()
157	slacapra	1.47	common.logger.debug(2, "Creation Time: "+str(stop - start))
158	slacapra	1.45	common.logger.write("Creation Time: "+str(stop - start))
159
160	nsmirnov	1.1	msg = '\nTotal of %d jobs created'%njc
161			if njc != self.ncjobs: msg = msg + ' from %d requested'%self.ncjobs
162			msg = msg + '.\n'
163			common.logger.message(msg)
164	slacapra	1.45
165	nsmirnov	1.1	return