CRAB/python/cms_orca.py

from JobType import JobType
from crab_logger import Logger
from crab_exceptions import *
from crab_util import *
import common
import PubDB
import orcarcBuilder
import orcarcBuilderOld
import Scram

import os, string, re

class Orca(JobType):
    def __init__(self, cfg_params):
        JobType.__init__(self, 'ORCA')
        common.logger.debug(3,'ORCA::__init__')

        self.analisys_common_info = {}

        log = common.logger
        
        self.scram = Scram.Scram(cfg_params)
        scramArea = ''
        self.additional_inbox_files = []
        self.scriptExe = ''

        self.version = self.scram.getSWVersion()
        common.analisys_common_info['sw_version'] = self.version

        ### collect Data cards
        try:
            self.owner = cfg_params['USER.owner']
            log.debug(6, "Orca::Orca(): owner = "+self.owner)
            self.dataset = cfg_params['USER.dataset']
            log.debug(6, "Orca::Orca(): dataset = "+self.dataset)
        except KeyError:
            msg = "Error: owner and/or dataset not defined "
            raise CrabException(msg)

        self.dataTiers = []
        try:
            tmpDataTiers = string.split(cfg_params['USER.data_tier'],',')
            for tmp in tmpDataTiers:
                tmp=string.strip(tmp)
                self.dataTiers.append(tmp)
                pass
            pass
        except KeyError:
            pass
        log.debug(6, "Orca::Orca(): dataTiers = "+str(self.dataTiers))

        ## now the application
        try:
            self.executable = cfg_params['USER.executable']
            log.debug(6, "Orca::Orca(): executable = "+self.executable)
        except KeyError:
            msg = "Error: executable not defined "
            raise CrabException(msg)

        try:
            self.orcarc_file = cfg_params['USER.orcarc_file']
            log.debug(6, "Orca::Orca(): orcarc file = "+self.orcarc_file)
            if (not os.path.exists(self.orcarc_file)):
                raise CrabException("User defined .orcarc file "+self.orcarc_file+" does not exist")
        except KeyError:
            log.message("Using empty orcarc file")
            self.orcarc_file = ''

        # output files
        try:
            self.output_file = []

            tmp = cfg_params['USER.output_file']
            if tmp != '':
                tmpOutFiles = string.split(cfg_params['USER.output_file'],',')
                log.debug(7, 'Orca::Orca(): output files '+str(tmpOutFiles))
                for tmp in tmpOutFiles:
                    tmp=string.strip(tmp)
                    self.output_file.append(tmp)
                    pass

            else:
                log.message("No output file defined: only stdout/err will be available")
                pass
            pass
        except KeyError:
            log.message("No output file defined: only stdout/err will be available")
            pass

        # script_exe file as additional file in inputSandbox
        try:
           self.scriptExe = cfg_params['USER.script_exe']
           self.additional_inbox_files.append(self.scriptExe)
        except KeyError:
           pass
        if self.scriptExe != '':
           if os.path.isfile(self.scriptExe):
              pass
           else:
              log.message("WARNING. file "+self.scriptExe+" not found")
              sys.exit()
                  
        ## additional input files
        try:
            tmpAddFiles = string.split(cfg_params['USER.additional_input_files'],',')
            for tmp in tmpAddFiles:
                tmp=string.strip(tmp)
                self.additional_inbox_files.append(tmp)
                pass
            pass
        except KeyError:
            pass

        try:
            self.total_number_of_events = int(cfg_params['USER.total_number_of_events'])
        except KeyError:
            msg = 'Must define total_number_of_events and job_number_of_events'
            raise CrabException(msg)

        try:
            self.first = int(cfg_params['USER.first_event'])
        except KeyError:
            self.first = 0
            pass
        log.debug(6, "Orca::Orca(): total number of events = "+`self.total_number_of_events`)
        #log.debug(6, "Orca::Orca(): events per job = "+`self.job_number_of_events`)
        log.debug(6, "Orca::Orca(): first event = "+`self.first`)
        
        self.maxEvents=0 # max events available in any PubDB
        self.connectPubDB(cfg_params)
          
        # [-- self.checkNevJobs() --]

        self.tgzNameWithPath = self.scram.getTarBall(self.executable)

        try:
            self.ML = int(cfg_params['USER.activate_monalisa'])
        except KeyError:
            self.ML = 0
            pass
        return

    def wsSetupEnvironment(self, nj):
        """
        Returns part of a job script which prepares
        the execution environment for the job 'nj'.
        """

        # Prepare JobType-independent part
        txt = self.wsSetupCMSEnvironment_()

        # Prepare JobType-specific part
        scram = self.scram.commandName()
        txt += '\n\n'
        txt += 'echo "### SPECIFIC JOB SETUP ENVIRONMENT ###"\n'
        txt += scram+' project ORCA '+self.version+'\n'
        txt += 'status=$?\n'
        txt += 'if [ $status != 0 ] ; then\n'
        txt += '   echo "SET_EXE_ENV 1 ==>ERROR ORCA '+self.version+' not found on `hostname`" \n'
        txt += '   echo "JOB_EXIT_STATUS = 5"\n'
        txt += '   echo "SanityCheckCode = 5" | tee -a $RUNTIME_AREA/$repo\n'
        txt += '   dumpStatus $RUNTIME_AREA/$repo\n'
        txt += '   exit 5 \n'
        txt += 'fi \n'
        txt += 'echo "ORCA_VERSION =  '+self.version+'"\n'
        txt += 'cd '+self.version+'\n'
        txt += 'eval `'+scram+' runtime -sh`\n'

        # Handle the arguments:
        txt += "\n"
        txt += "## ARGUMNETS: $1 Job Number\n"
        txt += "## ARGUMNETS: $2 First Event for this job\n"
        txt += "## ARGUMNETS: $3 Max Event for this job\n"
        txt += "\n"
        txt += "narg=$#\n"
        txt += "if [ $narg -lt 3 ]\n"
        txt += "then\n"
        txt += "    echo 'SET_EXE_ENV 1 ==> ERROR Too few arguments' +$narg+ \n"
        txt += '    echo "JOB_EXIT_STATUS = 1"\n'
        txt += '    echo "SanityCheckCode = 1" | tee -a $RUNTIME_AREA/$repo\n'
        txt += '    dumpStatus $RUNTIME_AREA/$repo\n'
        txt += "    exit 1\n"
        txt += "fi\n"
        txt += "\n"
        txt += "NJob=$1\n"
        txt += "FirstEvent=$2\n"
        txt += "MaxEvents=$3\n"

        # Prepare job-specific part
        job = common.job_list[nj]
        orcarc = os.path.basename(job.configFilename())
        txt += '\n'
        txt += 'cp $RUNTIME_AREA/'+orcarc+' .orcarc\n'
        txt += 'if [ -e $RUNTIME_AREA/orcarc_$CE ] ; then\n'
        txt += '  cat $RUNTIME_AREA/orcarc_$CE .orcarc >> .orcarc_tmp\n'
        txt += '  mv .orcarc_tmp .orcarc\n'
        txt += 'fi\n'
        txt += 'if [ -e $RUNTIME_AREA/init_$CE.sh ] ; then\n'
        txt += '  cp $RUNTIME_AREA/init_$CE.sh init.sh\n'
        txt += 'fi\n'

        if len(self.additional_inbox_files) > 0:
            for file in self.additional_inbox_files:
                txt += 'if [ -e $RUNTIME_AREA/'+file+' ] ; then\n'
                txt += '   cp $RUNTIME_AREA/'+file+' .\n'
                txt += '   chmod +x '+file+'\n'
                txt += 'fi\n'
            pass 

        txt += '\n'
        txt += 'chmod +x ./init.sh\n'
        txt += './init.sh\n'
        txt += 'exitStatus=$?\n'
        txt += 'if [ $exitStatus != 0 ] ; then\n'
        txt += '  echo "SET_EXE_ENV 1 ==> ERROR StageIn init script failed"\n'
        txt += '  echo "JOB_EXIT_STATUS = $exitStatus" \n'
        txt += '  echo "SanityCheckCode = $exitStatus" | tee -a $RUNTIME_AREA/$repo\n'
        txt += '  dumpStatus $RUNTIME_AREA/$repo\n'
        txt += '  exit $exitStatus\n'
        txt += 'fi\n'
        txt += "echo 'SET_EXE_ENV 0 ==> job setup ok'\n"
        txt += 'echo "### END JOB SETUP ENVIRONMENT ###"\n\n'

        txt += 'echo "FirstEvent=$FirstEvent" >> .orcarc\n'
        txt += 'echo "MaxEvents=$MaxEvents" >> .orcarc\n'
        if self.ML:
            txt += 'echo "MonalisaJobId=$NJob" >> .orcarc\n'

        txt += '\n'
        txt += 'echo "***** cat .orcarc *********"\n'
        txt += 'cat .orcarc\n'
        txt += 'echo "****** end .orcarc ********"\n'
        return txt
    
    def wsBuildExe(self, nj):
        """
        Put in the script the commands to build an executable
        or a library.
        """

        txt = ""

        if os.path.isfile(self.tgzNameWithPath):
            txt += 'echo "tar xzvf ../'+os.path.basename(self.tgzNameWithPath)+'"\n'
            txt += 'tar xzvf ../'+os.path.basename(self.tgzNameWithPath)+'\n'
            txt += 'untar_status=$? \n'
            txt += 'if [ $untar_status -ne 0 ]; then \n'
            txt += '   echo "SET_EXE 1 ==> ERROR Untarring .tgz file failed"\n'
            txt += '   echo "JOB_EXIT_STATUS = $untar_status" \n'
            txt += '   echo "SanityCheckCode = $untar_status" | tee -a $repo\n'
            txt += '   exit $untar_status \n'
            txt += 'else \n'
            txt += '   echo "Successful untar" \n'
            txt += 'fi \n'
            # TODO: what does this code do here ?
            # SL check that lib/Linux__... is present
            txt += 'mkdir -p lib/${SCRAM_ARCH} \n'
            txt += 'eval `'+self.scram.commandName()+' runtime -sh`'+'\n'
            pass

        return txt

    def wsRenameOutput(self, nj):
        """
        Returns part of a job script which renames the produced files.
        """

        txt = '\n'
        file_list = ''
        for fileWithSuffix in self.output_file:
            output_file_num = self.numberFile_(fileWithSuffix, '$NJob')
            file_list=file_list+output_file_num+','
            txt += '\n'
            txt += 'ls \n'
            txt += '\n'
            txt += 'ls '+fileWithSuffix+'\n'
            txt += 'exe_result=$?\n'
            txt += 'if [ $exe_result -ne 0 ] ; then\n'
            txt += '   echo "ERROR: No output file to manage"\n'
            txt += '   echo "JOB_EXIT_STATUS = $exe_result"\n'
            txt += '   echo "SanityCheckCode = $exe_result" | tee -a $RUNTIME_AREA/$repo\n'
            txt += '   dumpStatus $RUNTIME_AREA/$repo\n'
            txt += '   exit $exe_result \n'
            txt += 'else\n'
            txt += '   cp '+fileWithSuffix+' $RUNTIME_AREA/'+output_file_num+'\n'
            txt += 'fi\n'
            txt += 'cd $RUNTIME_AREA\n'
                      
            pass
       
        file_list=file_list[:-1]
        txt += 'file_list='+file_list+'\n'
        return txt

    def executableName(self):
        if self.scriptExe != '':
            return "./" + os.path.basename(self.scriptExe)
        else:
            return self.executable

    def connectPubDB(self, cfg_params):

        fun = "Orca::connectPubDB()"
        
        self.allOrcarcs = []
        # first check if the info from PubDB have been already processed
        if os.path.exists(common.work_space.shareDir()+'PubDBSummaryFile') :
            common.logger.debug(6, fun+": info from PubDB has been already processed -- use it")
            f = open( common.work_space.shareDir()+'PubDBSummaryFile', 'r' )
            for i in f.readlines():
                a=string.split(i,' ')
                self.allOrcarcs.append(orcarcBuilderOld.constructFromFile(a[0:-1]))
                pass
            for o in self.allOrcarcs:
                # o.dump()
                if o.Nevents >= self.maxEvents:
                    self.maxEvents= o.Nevents
                    pass
                pass
            pass

        else:  # PubDB never queried
            common.logger.debug(6, fun+": PubDB was never queried -- do it")
            # New PubDB class by SL
            try:
                self.pubdb = PubDB.PubDB(self.owner,
                                         self.dataset,
                                         self.dataTiers,
                                         cfg_params)
            except PubDB.RefDBmapError:
                msg = 'ERROR ***: accessing PubDB'
                raise CrabException(msg)

            ## extract info from pubDB (grouped by PubDB version :
            ##   pubDBData contains a list of info for the new-style PubDBs,
            ##   and a list of info for the old-style PubDBs )
            self.pubDBData = self.pubdb.getAllPubDBData()

            ## check and exit if no data are published in any PubDB
            nodata=1
            for PubDBversion in self.pubDBData.keys():
                if len(self.pubDBData[PubDBversion])>0:
                    nodata=0
            if (nodata):
                msg = 'Owner '+self.owner+' Dataset '+ self.dataset+ ' not published in any PubDB with asked dataTiers '+string.join(self.dataTiers,'-')+' ! '
                raise CrabException(msg)

           ## logging PubDB content for debugging 
            for PubDBversion in self.pubDBData.keys():
                common.logger.debug(6, fun+": PubDB "+PubDBversion+" info ("+`len(self.pubDBData[PubDBversion])`+"):\/")
                for aa in self.pubDBData[PubDBversion]:
                    common.logger.debug(6, "---------- start of a PubDB")
                    for bb in aa:
                        if common.logger.debugLevel() >= 6 :
                            common.logger.debug(6, str(bb.dump()))
                            pass
                        pass
                common.logger.debug(6, "----------- end of a PubDB")
            common.logger.debug(6, fun+": End of PubDB "+PubDBversion+" info\n")


            ## building orcarc : switch between info from old and new-style PubDB
            currDir = os.getcwd()
            os.chdir(common.work_space.jobDir())

            tmpOrcarcList=[]
            for PubDBversion in self.pubDBData.keys():
                if len(self.pubDBData[PubDBversion])>0 :
                    #print (" PubDB-style : %s"%(PubDBversion))
                    if PubDBversion=='newPubDB' :
                        self.builder = orcarcBuilder.orcarcBuilder(cfg_params)
                    else :
                        self.builder = orcarcBuilderOld.orcarcBuilderOld()
                    tmpAllOrcarcs = self.builder.createOrcarcAndInit(self.pubDBData[PubDBversion])
                    tmpOrcarcList.append(tmpAllOrcarcs)
                    #print 'version ',PubDBversion,' tmpAllOrcarcs ', tmpAllOrcarcs

            #print tmpOrcarcList
            os.chdir(currDir)

            self.maxEvents=0
            for tmpAllOrcarcs in tmpOrcarcList:
                for o in tmpAllOrcarcs:
                    numEvReq=self.total_number_of_events
                    if ((numEvReq == '-1') | (numEvReq <= o.Nevents)):
                        self.allOrcarcs.append(o)
                        if o.Nevents >= self.maxEvents:
                            self.maxEvents= o.Nevents
                            pass
                        pass
                    pass
          
            # set maximum number of event available

            # I save to a file self.allOrcarcs
          
            PubDBSummaryFile = open(common.work_space.shareDir()+'PubDBSummaryFile','w')
            for o in self.allOrcarcs:
                for d in o.content():
                    PubDBSummaryFile.write(d)
                    PubDBSummaryFile.write(' ')
                    pass
                PubDBSummaryFile.write('\n')
                pass
            PubDBSummaryFile.close()

            # for o in self.allOrcarcs:
            #     o.dump()
            pass

        # build a list of sites
        ces= []
        for o in self.allOrcarcs:
            ces.append(o.CE)
            pass

        if len(ces)==0:
            msg = 'No PubDBs publish correct catalogs or enough events! '
            msg += `self.total_number_of_events`
            raise CrabException(msg)

        common.logger.debug(6, "List of CEs: "+str(ces))
        common.analisys_common_info['sites']=ces

        return

    def nJobs(self):
        # TODO: should not be here !
        # JobType should have no internal knowledge about submitted jobs
        # One possibility is to use len(common.job_list).
        """ return the number of job to be created """
        return len(common.job_list)

    def prepareSteeringCards(self):
        """
        modify the orcarc card provided by the user, 
        writing a new card into share dir
        """
        infile = ''
        try:
          infile = open(self.orcarc_file,'r')
        except:
          self.orcarc_file = 'empty.orcarc'
          cmd='touch '+self.orcarc_file
          runCommand(cmd)
          infile = open(self.orcarc_file,'r')
            
        outfile = open(common.work_space.jobDir()+self.name()+'.orcarc', 'w')
           
        inline=infile.readlines()
        ### remove from user card these lines ###
        wordRemove=['InputFileCatalogURL', 'InputCollections', 'FirstEvent', 'MaxEvents', 'TFileAdaptor']
        for line in inline:
            word = string.strip(string.split(line,'=')[0])

            if word not in wordRemove:
                outfile.write(line)
            else:
                continue
            pass

        outfile.write('\n\n##### The following cards have been created by CRAB: DO NOT TOUCH #####\n')
        outfile.write('TFileAdaptor = true\n')

        if (self.ML) :
            outfile.write('MonalisaAddPid = false\n')
            outfile.write('ExtraPackages=RecApplPlugins\n')
            outfile.write('MonRecAlisaBuilder=true\n')
            ## TaskId is username+crab_0_date_time : that should be unique
            TaskID = os.getlogin()+'_'+string.split(common.work_space.topDir(),'/')[-2]
            outfile.write('MonalisaApplName='+TaskID+'\n')
            outfile.write('MonalisaNode=192.91.245.5\n')
            outfile.write('MonalisaPort=58884\n')
            pass

        outfile.write('InputCollections=/System/'+self.owner+'/'+self.dataset+'/'+self.dataset+'\n')

        infile.close()
        outfile.close()
        return
    
    def modifySteeringCards(self, nj):
        """
        Creates steering cards file modifying a template file
        """
        return

    def cardsBaseName(self):
        """
        Returns name of user orcarc card-file
        """
        return os.path.split (self.orcarc_file)[1]

    ### content of input_sanbdox ...
    def inputSandbox(self, nj):
        """
        Returns a list of filenames to be put in JDL input sandbox.
        """
        inp_box = []
        # dict added to delete duplicate from input sandbox file list
        seen = {}
        ## code
        if os.path.isfile(self.tgzNameWithPath):
            inp_box.append(self.tgzNameWithPath)
        ## orcarc
        for o in self.allOrcarcs:
          for f in o.fileList():
            if (f not in seen.keys()):
              inp_box.append(common.work_space.jobDir()+f)
              seen[f] = 1

        ## config
        inp_box.append(common.job_list[nj].configFilename())
        ## additional input files
        #inp_box = inp_box + self.additional_inbox_files
        #print "sono inputSandbox, inp_box = ", inp_box
        return inp_box

    ### and of output_sandbox
    def outputSandbox(self, nj):
        """
        Returns a list of filenames to be put in JDL output sandbox.
        """
        out_box = []

        stdout=common.job_list[nj].stdout()
        stderr=common.job_list[nj].stderr()
        #out_box.append(stdout)
        #out_box.append(stderr)

        ## User Declared output files
        for out in self.output_file:
            n_out = nj + 1 
            #FEDE 
            #out_box.append(self.version+'/'+self.numberFile_(out,str(n_out)))
            out_box.append(self.numberFile_(out,str(n_out)))
        return out_box

    def numberFile_(self, file, txt):
        """
        append _'txt' before last extension of a file
        """
        p = string.split(file,".")
        # take away last extension
        name = p[0]
        for x in p[1:-1]:
           name=name+"."+x
        # add "_txt"
        if len(p)>1:
          ext = p[len(p)-1]
          #result = name + '_' + str(txt) + "." + ext
          result = name + '_' + txt + "." + ext
        else:
          #result = name + '_' + str(txt)
          result = name + '_' + txt
        
        return result


    def stdOut(self):
        return self.stdOut_

    def stdErr(self):
        return self.stdErr_
Revision:	1.28
Committed:	Thu Dec 15 16:15:32 2005 UTC (19 years, 4 months ago) by corvo
Content type:	text/x-python
Branch:	MAIN
CVS Tags:	CRAB_1_0_4
Changes since 1.27:	+16 -7 lines
Log Message:	New 'production compliant' error code
#	User	Rev	Content
1	nsmirnov	1.1	from JobType import JobType
2			from crab_logger import Logger
3			from crab_exceptions import *
4			from crab_util import *
5			import common
6	nsmirnov	1.3	import PubDB
7	nsmirnov	1.1	import orcarcBuilder
8	slacapra	1.9	import orcarcBuilderOld
9			import Scram
10	nsmirnov	1.1
11			import os, string, re
12
13			class Orca(JobType):
14			def __init__(self, cfg_params):
15			JobType.__init__(self, 'ORCA')
16	slacapra	1.6	common.logger.debug(3,'ORCA::__init__')
17	nsmirnov	1.1
18			self.analisys_common_info = {}
19
20			log = common.logger
21
22	slacapra	1.9	self.scram = Scram.Scram(cfg_params)
23	nsmirnov	1.1	scramArea = ''
24	corvo	1.20	self.additional_inbox_files = []
25			self.scriptExe = ''
26	slacapra	1.6
27	slacapra	1.9	self.version = self.scram.getSWVersion()
28			common.analisys_common_info['sw_version'] = self.version
29	nsmirnov	1.1
30	slacapra	1.9	### collect Data cards
31	nsmirnov	1.1	try:
32			self.owner = cfg_params['USER.owner']
33	nsmirnov	1.3	log.debug(6, "Orca::Orca(): owner = "+self.owner)
34	nsmirnov	1.1	self.dataset = cfg_params['USER.dataset']
35	nsmirnov	1.3	log.debug(6, "Orca::Orca(): dataset = "+self.dataset)
36	slacapra	1.9	except KeyError:
37			msg = "Error: owner and/or dataset not defined "
38			raise CrabException(msg)
39
40			self.dataTiers = []
41			try:
42			tmpDataTiers = string.split(cfg_params['USER.data_tier'],',')
43			for tmp in tmpDataTiers:
44			tmp=string.strip(tmp)
45			self.dataTiers.append(tmp)
46			pass
47			pass
48			except KeyError:
49			pass
50			log.debug(6, "Orca::Orca(): dataTiers = "+str(self.dataTiers))
51
52			## now the application
53			try:
54	nsmirnov	1.1	self.executable = cfg_params['USER.executable']
55	nsmirnov	1.3	log.debug(6, "Orca::Orca(): executable = "+self.executable)
56	slacapra	1.23	except KeyError:
57	slacapra	1.9	msg = "Error: executable not defined "
58			raise CrabException(msg)
59
60			try:
61	nsmirnov	1.1	self.orcarc_file = cfg_params['USER.orcarc_file']
62	nsmirnov	1.3	log.debug(6, "Orca::Orca(): orcarc file = "+self.orcarc_file)
63	slacapra	1.23	if (not os.path.exists(self.orcarc_file)):
64			raise CrabException("User defined .orcarc file "+self.orcarc_file+" does not exist")
65			except KeyError:
66	slacapra	1.9	log.message("Using empty orcarc file")
67			self.orcarc_file = ''
68	nsmirnov	1.1
69	slacapra	1.9	# output files
70			try:
71	nsmirnov	1.1	self.output_file = []
72
73			tmp = cfg_params['USER.output_file']
74			if tmp != '':
75			tmpOutFiles = string.split(cfg_params['USER.output_file'],',')
76	nsmirnov	1.3	log.debug(7, 'Orca::Orca(): output files '+str(tmpOutFiles))
77	nsmirnov	1.1	for tmp in tmpOutFiles:
78			tmp=string.strip(tmp)
79			self.output_file.append(tmp)
80			pass
81
82	slacapra	1.9	else:
83			log.message("No output file defined: only stdout/err will be available")
84	nsmirnov	1.1	pass
85			pass
86			except KeyError:
87	slacapra	1.9	log.message("No output file defined: only stdout/err will be available")
88	nsmirnov	1.1	pass
89
90	corvo	1.20	# script_exe file as additional file in inputSandbox
91			try:
92			self.scriptExe = cfg_params['USER.script_exe']
93			self.additional_inbox_files.append(self.scriptExe)
94			except KeyError:
95			pass
96			if self.scriptExe != '':
97			if os.path.isfile(self.scriptExe):
98			pass
99			else:
100			log.message("WARNING. file "+self.scriptExe+" not found")
101			sys.exit()
102
103	slacapra	1.9	## additional input files
104	nsmirnov	1.1	try:
105			tmpAddFiles = string.split(cfg_params['USER.additional_input_files'],',')
106			for tmp in tmpAddFiles:
107			tmp=string.strip(tmp)
108	slacapra	1.7	self.additional_inbox_files.append(tmp)
109	nsmirnov	1.1	pass
110			pass
111			except KeyError:
112			pass
113
114			try:
115			self.total_number_of_events = int(cfg_params['USER.total_number_of_events'])
116			except KeyError:
117			msg = 'Must define total_number_of_events and job_number_of_events'
118			raise CrabException(msg)
119
120			try:
121			self.first = int(cfg_params['USER.first_event'])
122			except KeyError:
123			self.first = 0
124			pass
125	nsmirnov	1.3	log.debug(6, "Orca::Orca(): total number of events = "+`self.total_number_of_events`)
126	slacapra	1.6	#log.debug(6, "Orca::Orca(): events per job = "+`self.job_number_of_events`)
127	nsmirnov	1.3	log.debug(6, "Orca::Orca(): first event = "+`self.first`)
128	nsmirnov	1.1
129			self.maxEvents=0 # max events available in any PubDB
130	slacapra	1.7	self.connectPubDB(cfg_params)
131	nsmirnov	1.1
132	slacapra	1.9	# [-- self.checkNevJobs() --]
133	nsmirnov	1.1
134	slacapra	1.9	self.tgzNameWithPath = self.scram.getTarBall(self.executable)
135	nsmirnov	1.1
136	slacapra	1.10	try:
137			self.ML = int(cfg_params['USER.activate_monalisa'])
138			except KeyError:
139			self.ML = 0
140			pass
141	nsmirnov	1.1	return
142
143	nsmirnov	1.4	def wsSetupEnvironment(self, nj):
144			"""
145			Returns part of a job script which prepares
146			the execution environment for the job 'nj'.
147			"""
148
149			# Prepare JobType-independent part
150			txt = self.wsSetupCMSEnvironment_()
151
152	fanzago	1.16	# Prepare JobType-specific part
153			scram = self.scram.commandName()
154			txt += '\n\n'
155			txt += 'echo "### SPECIFIC JOB SETUP ENVIRONMENT ###"\n'
156			txt += scram+' project ORCA '+self.version+'\n'
157			txt += 'status=$?\n'
158			txt += 'if [ $status != 0 ] ; then\n'
159			txt += ' echo "SET_EXE_ENV 1 ==>ERROR ORCA '+self.version+' not found on `hostname`" \n'
160	corvo	1.28	txt += ' echo "JOB_EXIT_STATUS = 5"\n'
161			txt += ' echo "SanityCheckCode = 5" \| tee -a $RUNTIME_AREA/$repo\n'
162			txt += ' dumpStatus $RUNTIME_AREA/$repo\n'
163			txt += ' exit 5 \n'
164	fanzago	1.16	txt += 'fi \n'
165			txt += 'echo "ORCA_VERSION = '+self.version+'"\n'
166			txt += 'cd '+self.version+'\n'
167			txt += 'eval `'+scram+' runtime -sh`\n'
168
169	slacapra	1.9	# Handle the arguments:
170			txt += "\n"
171			txt += "## ARGUMNETS: $1 Job Number\n"
172			txt += "## ARGUMNETS: $2 First Event for this job\n"
173			txt += "## ARGUMNETS: $3 Max Event for this job\n"
174			txt += "\n"
175			txt += "narg=$#\n"
176			txt += "if [ $narg -lt 3 ]\n"
177			txt += "then\n"
178	fanzago	1.16	txt += " echo 'SET_EXE_ENV 1 ==> ERROR Too few arguments' +$narg+ \n"
179	spiga	1.17	txt += ' echo "JOB_EXIT_STATUS = 1"\n'
180	corvo	1.28	txt += ' echo "SanityCheckCode = 1" \| tee -a $RUNTIME_AREA/$repo\n'
181			txt += ' dumpStatus $RUNTIME_AREA/$repo\n'
182	slacapra	1.9	txt += " exit 1\n"
183			txt += "fi\n"
184	fanzago	1.16	txt += "\n"
185	slacapra	1.10	txt += "NJob=$1\n"
186			txt += "FirstEvent=$2\n"
187	afanfani	1.24	txt += "MaxEvents=$3\n"
188	nsmirnov	1.4
189			# Prepare job-specific part
190			job = common.job_list[nj]
191			orcarc = os.path.basename(job.configFilename())
192			txt += '\n'
193			txt += 'cp $RUNTIME_AREA/'+orcarc+' .orcarc\n'
194	slacapra	1.22	txt += 'if [ -e $RUNTIME_AREA/orcarc_$CE ] ; then\n'
195	corvo	1.21	txt += ' cat $RUNTIME_AREA/orcarc_$CE .orcarc >> .orcarc_tmp\n'
196	nsmirnov	1.4	txt += ' mv .orcarc_tmp .orcarc\n'
197	slacapra	1.22	txt += 'fi\n'
198			txt += 'if [ -e $RUNTIME_AREA/init_$CE.sh ] ; then\n'
199	corvo	1.21	txt += ' cp $RUNTIME_AREA/init_$CE.sh init.sh\n'
200	nsmirnov	1.4	txt += 'fi\n'
201	fanzago	1.26
202			if len(self.additional_inbox_files) > 0:
203			for file in self.additional_inbox_files:
204			txt += 'if [ -e $RUNTIME_AREA/'+file+' ] ; then\n'
205			txt += ' cp $RUNTIME_AREA/'+file+' .\n'
206	fanzago	1.27	txt += ' chmod +x '+file+'\n'
207	fanzago	1.26	txt += 'fi\n'
208			pass
209
210	nsmirnov	1.4	txt += '\n'
211			txt += 'chmod +x ./init.sh\n'
212			txt += './init.sh\n'
213			txt += 'exitStatus=$?\n'
214			txt += 'if [ $exitStatus != 0 ] ; then\n'
215	fanzago	1.16	txt += ' echo "SET_EXE_ENV 1 ==> ERROR StageIn init script failed"\n'
216	corvo	1.28	txt += ' echo "JOB_EXIT_STATUS = $exitStatus" \n'
217			txt += ' echo "SanityCheckCode = $exitStatus" \| tee -a $RUNTIME_AREA/$repo\n'
218			txt += ' dumpStatus $RUNTIME_AREA/$repo\n'
219	nsmirnov	1.4	txt += ' exit $exitStatus\n'
220			txt += 'fi\n'
221	fanzago	1.16	txt += "echo 'SET_EXE_ENV 0 ==> job setup ok'\n"
222			txt += 'echo "### END JOB SETUP ENVIRONMENT ###"\n\n'
223	slacapra	1.9
224	slacapra	1.10	txt += 'echo "FirstEvent=$FirstEvent" >> .orcarc\n'
225	afanfani	1.24	txt += 'echo "MaxEvents=$MaxEvents" >> .orcarc\n'
226	slacapra	1.10	if self.ML:
227			txt += 'echo "MonalisaJobId=$NJob" >> .orcarc\n'
228
229			txt += '\n'
230			txt += 'echo "*** cat .orcarc *******"\n'
231			txt += 'cat .orcarc\n'
232			txt += 'echo "**** end .orcarc ******"\n'
233	nsmirnov	1.4	return txt
234
235			def wsBuildExe(self, nj):
236			"""
237			Put in the script the commands to build an executable
238			or a library.
239			"""
240
241			txt = ""
242
243	fanzago	1.12	if os.path.isfile(self.tgzNameWithPath):
244			txt += 'echo "tar xzvf ../'+os.path.basename(self.tgzNameWithPath)+'"\n'
245			txt += 'tar xzvf ../'+os.path.basename(self.tgzNameWithPath)+'\n'
246	nsmirnov	1.4	txt += 'untar_status=$? \n'
247			txt += 'if [ $untar_status -ne 0 ]; then \n'
248	fanzago	1.18	txt += ' echo "SET_EXE 1 ==> ERROR Untarring .tgz file failed"\n'
249	corvo	1.28	txt += ' echo "JOB_EXIT_STATUS = $untar_status" \n'
250			txt += ' echo "SanityCheckCode = $untar_status" \| tee -a $repo\n'
251			txt += ' exit $untar_status \n'
252	nsmirnov	1.4	txt += 'else \n'
253			txt += ' echo "Successful untar" \n'
254			txt += 'fi \n'
255			# TODO: what does this code do here ?
256			# SL check that lib/Linux__... is present
257	slacapra	1.9	txt += 'mkdir -p lib/${SCRAM_ARCH} \n'
258			txt += 'eval `'+self.scram.commandName()+' runtime -sh`'+'\n'
259	nsmirnov	1.4	pass
260
261			return txt
262
263			def wsRenameOutput(self, nj):
264			"""
265			Returns part of a job script which renames the produced files.
266			"""
267	slacapra	1.9
268	nsmirnov	1.4	txt = '\n'
269	fanzago	1.15	file_list = ''
270	slacapra	1.9	for fileWithSuffix in self.output_file:
271			output_file_num = self.numberFile_(fileWithSuffix, '$NJob')
272	fanzago	1.15	file_list=file_list+output_file_num+','
273	fanzago	1.26	txt += '\n'
274			txt += 'ls \n'
275			txt += '\n'
276	spiga	1.17	txt += 'ls '+fileWithSuffix+'\n'
277	fanzago	1.18	txt += 'exe_result=$?\n'
278			txt += 'if [ $exe_result -ne 0 ] ; then\n'
279			txt += ' echo "ERROR: No output file to manage"\n'
280	corvo	1.28	txt += ' echo "JOB_EXIT_STATUS = $exe_result"\n'
281			txt += ' echo "SanityCheckCode = $exe_result" \| tee -a $RUNTIME_AREA/$repo\n'
282			txt += ' dumpStatus $RUNTIME_AREA/$repo\n'
283			txt += ' exit $exe_result \n'
284	spiga	1.17	txt += 'else\n'
285	fanzago	1.25	txt += ' cp '+fileWithSuffix+' $RUNTIME_AREA/'+output_file_num+'\n'
286			txt += 'fi\n'
287			txt += 'cd $RUNTIME_AREA\n'
288
289	nsmirnov	1.4	pass
290	spiga	1.17
291	fanzago	1.15	file_list=file_list[:-1]
292			txt += 'file_list='+file_list+'\n'
293	nsmirnov	1.4	return txt
294
295	nsmirnov	1.1	def executableName(self):
296	corvo	1.20	if self.scriptExe != '':
297			return "./" + os.path.basename(self.scriptExe)
298			else:
299			return self.executable
300	nsmirnov	1.1
301	slacapra	1.7	def connectPubDB(self, cfg_params):
302	nsmirnov	1.1
303	nsmirnov	1.3	fun = "Orca::connectPubDB()"
304
305	nsmirnov	1.1	self.allOrcarcs = []
306			# first check if the info from PubDB have been already processed
307			if os.path.exists(common.work_space.shareDir()+'PubDBSummaryFile') :
308	nsmirnov	1.3	common.logger.debug(6, fun+": info from PubDB has been already processed -- use it")
309	nsmirnov	1.1	f = open( common.work_space.shareDir()+'PubDBSummaryFile', 'r' )
310			for i in f.readlines():
311			a=string.split(i,' ')
312	slacapra	1.9	self.allOrcarcs.append(orcarcBuilderOld.constructFromFile(a[0:-1]))
313	nsmirnov	1.1	pass
314			for o in self.allOrcarcs:
315			# o.dump()
316			if o.Nevents >= self.maxEvents:
317			self.maxEvents= o.Nevents
318			pass
319			pass
320			pass
321
322			else: # PubDB never queried
323	nsmirnov	1.3	common.logger.debug(6, fun+": PubDB was never queried -- do it")
324	nsmirnov	1.1	# New PubDB class by SL
325			try:
326	nsmirnov	1.3	self.pubdb = PubDB.PubDB(self.owner,
327	nsmirnov	1.1	self.dataset,
328	slacapra	1.7	self.dataTiers,
329			cfg_params)
330	slacapra	1.9	except PubDB.RefDBmapError:
331	nsmirnov	1.1	msg = 'ERROR ***: accessing PubDB'
332			raise CrabException(msg)
333
334	slacapra	1.9	## extract info from pubDB (grouped by PubDB version :
335			## pubDBData contains a list of info for the new-style PubDBs,
336			## and a list of info for the old-style PubDBs )
337			self.pubDBData = self.pubdb.getAllPubDBData()
338
339			## check and exit if no data are published in any PubDB
340			nodata=1
341			for PubDBversion in self.pubDBData.keys():
342			if len(self.pubDBData[PubDBversion])>0:
343			nodata=0
344			if (nodata):
345			msg = 'Owner '+self.owner+' Dataset '+ self.dataset+ ' not published in any PubDB with asked dataTiers '+string.join(self.dataTiers,'-')+' ! '
346	nsmirnov	1.1	raise CrabException(msg)
347	nsmirnov	1.3
348	slacapra	1.9	## logging PubDB content for debugging
349			for PubDBversion in self.pubDBData.keys():
350			common.logger.debug(6, fun+": PubDB "+PubDBversion+" info ("+`len(self.pubDBData[PubDBversion])`+"):\/")
351			for aa in self.pubDBData[PubDBversion]:
352			common.logger.debug(6, "---------- start of a PubDB")
353			for bb in aa:
354			if common.logger.debugLevel() >= 6 :
355			common.logger.debug(6, str(bb.dump()))
356			pass
357			pass
358			common.logger.debug(6, "----------- end of a PubDB")
359			common.logger.debug(6, fun+": End of PubDB "+PubDBversion+" info\n")
360	nsmirnov	1.1
361
362	slacapra	1.9	## building orcarc : switch between info from old and new-style PubDB
363	nsmirnov	1.1	currDir = os.getcwd()
364			os.chdir(common.work_space.jobDir())
365	slacapra	1.9
366			tmpOrcarcList=[]
367			for PubDBversion in self.pubDBData.keys():
368			if len(self.pubDBData[PubDBversion])>0 :
369			#print (" PubDB-style : %s"%(PubDBversion))
370			if PubDBversion=='newPubDB' :
371	slacapra	1.19	self.builder = orcarcBuilder.orcarcBuilder(cfg_params)
372	slacapra	1.9	else :
373			self.builder = orcarcBuilderOld.orcarcBuilderOld()
374			tmpAllOrcarcs = self.builder.createOrcarcAndInit(self.pubDBData[PubDBversion])
375			tmpOrcarcList.append(tmpAllOrcarcs)
376			#print 'version ',PubDBversion,' tmpAllOrcarcs ', tmpAllOrcarcs
377
378			#print tmpOrcarcList
379	nsmirnov	1.1	os.chdir(currDir)
380
381			self.maxEvents=0
382	slacapra	1.9	for tmpAllOrcarcs in tmpOrcarcList:
383			for o in tmpAllOrcarcs:
384			numEvReq=self.total_number_of_events
385			if ((numEvReq == '-1') \| (numEvReq <= o.Nevents)):
386			self.allOrcarcs.append(o)
387			if o.Nevents >= self.maxEvents:
388			self.maxEvents= o.Nevents
389			pass
390	nsmirnov	1.1	pass
391			pass
392
393			# set maximum number of event available
394
395			# I save to a file self.allOrcarcs
396
397			PubDBSummaryFile = open(common.work_space.shareDir()+'PubDBSummaryFile','w')
398			for o in self.allOrcarcs:
399			for d in o.content():
400			PubDBSummaryFile.write(d)
401			PubDBSummaryFile.write(' ')
402			pass
403			PubDBSummaryFile.write('\n')
404			pass
405			PubDBSummaryFile.close()
406
407			# for o in self.allOrcarcs:
408	slacapra	1.7	# o.dump()
409	nsmirnov	1.1	pass
410
411			# build a list of sites
412			ces= []
413			for o in self.allOrcarcs:
414			ces.append(o.CE)
415			pass
416
417			if len(ces)==0:
418	slacapra	1.9	msg = 'No PubDBs publish correct catalogs or enough events! '
419	nsmirnov	1.1	msg += `self.total_number_of_events`
420			raise CrabException(msg)
421
422	nsmirnov	1.3	common.logger.debug(6, "List of CEs: "+str(ces))
423	slacapra	1.6	common.analisys_common_info['sites']=ces
424	nsmirnov	1.1
425			return
426
427			def nJobs(self):
428			# TODO: should not be here !
429			# JobType should have no internal knowledge about submitted jobs
430			# One possibility is to use len(common.job_list).
431			""" return the number of job to be created """
432	slacapra	1.6	return len(common.job_list)
433	nsmirnov	1.5
434			def prepareSteeringCards(self):
435			"""
436			modify the orcarc card provided by the user,
437			writing a new card into share dir
438			"""
439			infile = ''
440			try:
441			infile = open(self.orcarc_file,'r')
442			except:
443			self.orcarc_file = 'empty.orcarc'
444			cmd='touch '+self.orcarc_file
445	slacapra	1.9	runCommand(cmd)
446	nsmirnov	1.5	infile = open(self.orcarc_file,'r')
447
448	slacapra	1.9	outfile = open(common.work_space.jobDir()+self.name()+'.orcarc', 'w')
449	nsmirnov	1.5
450			inline=infile.readlines()
451			### remove from user card these lines ###
452	slacapra	1.9	wordRemove=['InputFileCatalogURL', 'InputCollections', 'FirstEvent', 'MaxEvents', 'TFileAdaptor']
453			for line in inline:
454			word = string.strip(string.split(line,'=')[0])
455	slacapra	1.8
456	slacapra	1.9	if word not in wordRemove:
457			outfile.write(line)
458			else:
459			continue
460			pass
461
462			outfile.write('\n\n##### The following cards have been created by CRAB: DO NOT TOUCH #####\n')
463			outfile.write('TFileAdaptor = true\n')
464
465	slacapra	1.10	if (self.ML) :
466			outfile.write('MonalisaAddPid = false\n')
467			outfile.write('ExtraPackages=RecApplPlugins\n')
468			outfile.write('MonRecAlisaBuilder=true\n')
469			## TaskId is username+crab_0_date_time : that should be unique
470			TaskID = os.getlogin()+'_'+string.split(common.work_space.topDir(),'/')[-2]
471			outfile.write('MonalisaApplName='+TaskID+'\n')
472			outfile.write('MonalisaNode=192.91.245.5\n')
473			outfile.write('MonalisaPort=58884\n')
474			pass
475
476	slacapra	1.9	outfile.write('InputCollections=/System/'+self.owner+'/'+self.dataset+'/'+self.dataset+'\n')
477	slacapra	1.8
478	nsmirnov	1.5	infile.close()
479			outfile.close()
480			return
481	nsmirnov	1.1
482			def modifySteeringCards(self, nj):
483			"""
484			Creates steering cards file modifying a template file
485			"""
486			return
487
488			def cardsBaseName(self):
489			"""
490			Returns name of user orcarc card-file
491			"""
492			return os.path.split (self.orcarc_file)[1]
493
494	fanzago	1.27	### content of input_sanbdox ...
495	nsmirnov	1.1	def inputSandbox(self, nj):
496			"""
497			Returns a list of filenames to be put in JDL input sandbox.
498			"""
499			inp_box = []
500	corvo	1.13	# dict added to delete duplicate from input sandbox file list
501			seen = {}
502	slacapra	1.7	## code
503	fanzago	1.12	if os.path.isfile(self.tgzNameWithPath):
504			inp_box.append(self.tgzNameWithPath)
505	slacapra	1.7	## orcarc
506	nsmirnov	1.1	for o in self.allOrcarcs:
507			for f in o.fileList():
508	corvo	1.13	if (f not in seen.keys()):
509			inp_box.append(common.work_space.jobDir()+f)
510			seen[f] = 1
511	slacapra	1.9
512	slacapra	1.7	## config
513	nsmirnov	1.1	inp_box.append(common.job_list[nj].configFilename())
514	slacapra	1.7	## additional input files
515	fanzago	1.26	#inp_box = inp_box + self.additional_inbox_files
516			#print "sono inputSandbox, inp_box = ", inp_box
517	nsmirnov	1.1	return inp_box
518
519	fanzago	1.15	### and of output_sandbox
520	nsmirnov	1.1	def outputSandbox(self, nj):
521			"""
522			Returns a list of filenames to be put in JDL output sandbox.
523			"""
524			out_box = []
525
526	slacapra	1.9	stdout=common.job_list[nj].stdout()
527			stderr=common.job_list[nj].stderr()
528	fanzago	1.15	#out_box.append(stdout)
529			#out_box.append(stderr)
530	slacapra	1.9
531	slacapra	1.7	## User Declared output files
532	slacapra	1.9	for out in self.output_file:
533	fanzago	1.14	n_out = nj + 1
534	fanzago	1.25	#FEDE
535			#out_box.append(self.version+'/'+self.numberFile_(out,str(n_out)))
536			out_box.append(self.numberFile_(out,str(n_out)))
537	nsmirnov	1.1	return out_box
538	slacapra	1.7
539	slacapra	1.9	def numberFile_(self, file, txt):
540			"""
541			append _'txt' before last extension of a file
542			"""
543			p = string.split(file,".")
544			# take away last extension
545			name = p[0]
546			for x in p[1:-1]:
547			name=name+"."+x
548			# add "_txt"
549			if len(p)>1:
550			ext = p[len(p)-1]
551	fanzago	1.14	#result = name + '_' + str(txt) + "." + ext
552			result = name + '_' + txt + "." + ext
553	slacapra	1.9	else:
554	fanzago	1.14	#result = name + '_' + str(txt)
555			result = name + '_' + txt
556	slacapra	1.9
557			return result
558
559
560	slacapra	1.7	def stdOut(self):
561			return self.stdOut_
562
563			def stdErr(self):
564			return self.stdErr_