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'
        ### needed grep for bug in scramv1 ###
        txt += 'eval `'+scram+' runtime -sh | grep -v SCRAMRT_LSB_JOBNAME`\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 (int(o.Nevents) >= self.maxEvents):
                            self.maxEvents= int(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()
            ### fede
            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.31
Committed:	Tue Feb 14 18:26:19 2006 UTC (19 years, 2 months ago) by slacapra
Content type:	text/x-python
Branch:	MAIN
Changes since 1.30:	+2 -2 lines
Log Message:	comparing int with int, not string grrrrr
#	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	fanzago	1.29	### needed grep for bug in scramv1 ###
168	fanzago	1.30	txt += 'eval `'+scram+' runtime -sh \| grep -v SCRAMRT_LSB_JOBNAME`\n'
169	fanzago	1.16
170	slacapra	1.9	# Handle the arguments:
171			txt += "\n"
172			txt += "## ARGUMNETS: $1 Job Number\n"
173			txt += "## ARGUMNETS: $2 First Event for this job\n"
174			txt += "## ARGUMNETS: $3 Max Event for this job\n"
175			txt += "\n"
176			txt += "narg=$#\n"
177			txt += "if [ $narg -lt 3 ]\n"
178			txt += "then\n"
179	fanzago	1.16	txt += " echo 'SET_EXE_ENV 1 ==> ERROR Too few arguments' +$narg+ \n"
180	spiga	1.17	txt += ' echo "JOB_EXIT_STATUS = 1"\n'
181	corvo	1.28	txt += ' echo "SanityCheckCode = 1" \| tee -a $RUNTIME_AREA/$repo\n'
182			txt += ' dumpStatus $RUNTIME_AREA/$repo\n'
183	slacapra	1.9	txt += " exit 1\n"
184			txt += "fi\n"
185	fanzago	1.16	txt += "\n"
186	slacapra	1.10	txt += "NJob=$1\n"
187			txt += "FirstEvent=$2\n"
188	afanfani	1.24	txt += "MaxEvents=$3\n"
189	nsmirnov	1.4
190			# Prepare job-specific part
191			job = common.job_list[nj]
192			orcarc = os.path.basename(job.configFilename())
193			txt += '\n'
194			txt += 'cp $RUNTIME_AREA/'+orcarc+' .orcarc\n'
195	slacapra	1.22	txt += 'if [ -e $RUNTIME_AREA/orcarc_$CE ] ; then\n'
196	corvo	1.21	txt += ' cat $RUNTIME_AREA/orcarc_$CE .orcarc >> .orcarc_tmp\n'
197	nsmirnov	1.4	txt += ' mv .orcarc_tmp .orcarc\n'
198	slacapra	1.22	txt += 'fi\n'
199			txt += 'if [ -e $RUNTIME_AREA/init_$CE.sh ] ; then\n'
200	corvo	1.21	txt += ' cp $RUNTIME_AREA/init_$CE.sh init.sh\n'
201	nsmirnov	1.4	txt += 'fi\n'
202	fanzago	1.26
203			if len(self.additional_inbox_files) > 0:
204			for file in self.additional_inbox_files:
205			txt += 'if [ -e $RUNTIME_AREA/'+file+' ] ; then\n'
206			txt += ' cp $RUNTIME_AREA/'+file+' .\n'
207	fanzago	1.27	txt += ' chmod +x '+file+'\n'
208	fanzago	1.26	txt += 'fi\n'
209			pass
210
211	nsmirnov	1.4	txt += '\n'
212			txt += 'chmod +x ./init.sh\n'
213			txt += './init.sh\n'
214			txt += 'exitStatus=$?\n'
215			txt += 'if [ $exitStatus != 0 ] ; then\n'
216	fanzago	1.16	txt += ' echo "SET_EXE_ENV 1 ==> ERROR StageIn init script failed"\n'
217	corvo	1.28	txt += ' echo "JOB_EXIT_STATUS = $exitStatus" \n'
218			txt += ' echo "SanityCheckCode = $exitStatus" \| tee -a $RUNTIME_AREA/$repo\n'
219			txt += ' dumpStatus $RUNTIME_AREA/$repo\n'
220	nsmirnov	1.4	txt += ' exit $exitStatus\n'
221			txt += 'fi\n'
222	fanzago	1.16	txt += "echo 'SET_EXE_ENV 0 ==> job setup ok'\n"
223			txt += 'echo "### END JOB SETUP ENVIRONMENT ###"\n\n'
224	slacapra	1.9
225	slacapra	1.10	txt += 'echo "FirstEvent=$FirstEvent" >> .orcarc\n'
226	afanfani	1.24	txt += 'echo "MaxEvents=$MaxEvents" >> .orcarc\n'
227	slacapra	1.10	if self.ML:
228			txt += 'echo "MonalisaJobId=$NJob" >> .orcarc\n'
229
230			txt += '\n'
231			txt += 'echo "*** cat .orcarc *******"\n'
232			txt += 'cat .orcarc\n'
233			txt += 'echo "**** end .orcarc ******"\n'
234	nsmirnov	1.4	return txt
235
236			def wsBuildExe(self, nj):
237			"""
238			Put in the script the commands to build an executable
239			or a library.
240			"""
241
242			txt = ""
243
244	fanzago	1.12	if os.path.isfile(self.tgzNameWithPath):
245			txt += 'echo "tar xzvf ../'+os.path.basename(self.tgzNameWithPath)+'"\n'
246			txt += 'tar xzvf ../'+os.path.basename(self.tgzNameWithPath)+'\n'
247	nsmirnov	1.4	txt += 'untar_status=$? \n'
248			txt += 'if [ $untar_status -ne 0 ]; then \n'
249	fanzago	1.18	txt += ' echo "SET_EXE 1 ==> ERROR Untarring .tgz file failed"\n'
250	corvo	1.28	txt += ' echo "JOB_EXIT_STATUS = $untar_status" \n'
251			txt += ' echo "SanityCheckCode = $untar_status" \| tee -a $repo\n'
252			txt += ' exit $untar_status \n'
253	nsmirnov	1.4	txt += 'else \n'
254			txt += ' echo "Successful untar" \n'
255			txt += 'fi \n'
256			# TODO: what does this code do here ?
257			# SL check that lib/Linux__... is present
258	slacapra	1.9	txt += 'mkdir -p lib/${SCRAM_ARCH} \n'
259			txt += 'eval `'+self.scram.commandName()+' runtime -sh`'+'\n'
260	nsmirnov	1.4	pass
261
262			return txt
263
264			def wsRenameOutput(self, nj):
265			"""
266			Returns part of a job script which renames the produced files.
267			"""
268	slacapra	1.9
269	nsmirnov	1.4	txt = '\n'
270	fanzago	1.15	file_list = ''
271	slacapra	1.9	for fileWithSuffix in self.output_file:
272			output_file_num = self.numberFile_(fileWithSuffix, '$NJob')
273	fanzago	1.15	file_list=file_list+output_file_num+','
274	fanzago	1.26	txt += '\n'
275			txt += 'ls \n'
276			txt += '\n'
277	spiga	1.17	txt += 'ls '+fileWithSuffix+'\n'
278	fanzago	1.18	txt += 'exe_result=$?\n'
279			txt += 'if [ $exe_result -ne 0 ] ; then\n'
280			txt += ' echo "ERROR: No output file to manage"\n'
281	corvo	1.28	txt += ' echo "JOB_EXIT_STATUS = $exe_result"\n'
282			txt += ' echo "SanityCheckCode = $exe_result" \| tee -a $RUNTIME_AREA/$repo\n'
283			txt += ' dumpStatus $RUNTIME_AREA/$repo\n'
284			txt += ' exit $exe_result \n'
285	spiga	1.17	txt += 'else\n'
286	fanzago	1.25	txt += ' cp '+fileWithSuffix+' $RUNTIME_AREA/'+output_file_num+'\n'
287			txt += 'fi\n'
288			txt += 'cd $RUNTIME_AREA\n'
289
290	nsmirnov	1.4	pass
291	spiga	1.17
292	fanzago	1.15	file_list=file_list[:-1]
293			txt += 'file_list='+file_list+'\n'
294	nsmirnov	1.4	return txt
295
296	nsmirnov	1.1	def executableName(self):
297	corvo	1.20	if self.scriptExe != '':
298			return "./" + os.path.basename(self.scriptExe)
299			else:
300			return self.executable
301	nsmirnov	1.1
302	slacapra	1.7	def connectPubDB(self, cfg_params):
303	nsmirnov	1.1
304	nsmirnov	1.3	fun = "Orca::connectPubDB()"
305
306	nsmirnov	1.1	self.allOrcarcs = []
307			# first check if the info from PubDB have been already processed
308			if os.path.exists(common.work_space.shareDir()+'PubDBSummaryFile') :
309	nsmirnov	1.3	common.logger.debug(6, fun+": info from PubDB has been already processed -- use it")
310	nsmirnov	1.1	f = open( common.work_space.shareDir()+'PubDBSummaryFile', 'r' )
311			for i in f.readlines():
312			a=string.split(i,' ')
313	slacapra	1.9	self.allOrcarcs.append(orcarcBuilderOld.constructFromFile(a[0:-1]))
314	nsmirnov	1.1	pass
315			for o in self.allOrcarcs:
316			# o.dump()
317			if o.Nevents >= self.maxEvents:
318			self.maxEvents= o.Nevents
319			pass
320			pass
321			pass
322
323			else: # PubDB never queried
324	nsmirnov	1.3	common.logger.debug(6, fun+": PubDB was never queried -- do it")
325	nsmirnov	1.1	# New PubDB class by SL
326			try:
327	nsmirnov	1.3	self.pubdb = PubDB.PubDB(self.owner,
328	nsmirnov	1.1	self.dataset,
329	slacapra	1.7	self.dataTiers,
330			cfg_params)
331	slacapra	1.9	except PubDB.RefDBmapError:
332	nsmirnov	1.1	msg = 'ERROR ***: accessing PubDB'
333			raise CrabException(msg)
334
335	slacapra	1.9	## extract info from pubDB (grouped by PubDB version :
336			## pubDBData contains a list of info for the new-style PubDBs,
337			## and a list of info for the old-style PubDBs )
338			self.pubDBData = self.pubdb.getAllPubDBData()
339
340			## check and exit if no data are published in any PubDB
341			nodata=1
342			for PubDBversion in self.pubDBData.keys():
343			if len(self.pubDBData[PubDBversion])>0:
344			nodata=0
345			if (nodata):
346			msg = 'Owner '+self.owner+' Dataset '+ self.dataset+ ' not published in any PubDB with asked dataTiers '+string.join(self.dataTiers,'-')+' ! '
347	nsmirnov	1.1	raise CrabException(msg)
348	nsmirnov	1.3
349	slacapra	1.9	## logging PubDB content for debugging
350			for PubDBversion in self.pubDBData.keys():
351			common.logger.debug(6, fun+": PubDB "+PubDBversion+" info ("+`len(self.pubDBData[PubDBversion])`+"):\/")
352			for aa in self.pubDBData[PubDBversion]:
353			common.logger.debug(6, "---------- start of a PubDB")
354			for bb in aa:
355			if common.logger.debugLevel() >= 6 :
356			common.logger.debug(6, str(bb.dump()))
357			pass
358			pass
359			common.logger.debug(6, "----------- end of a PubDB")
360			common.logger.debug(6, fun+": End of PubDB "+PubDBversion+" info\n")
361	nsmirnov	1.1
362
363	slacapra	1.9	## building orcarc : switch between info from old and new-style PubDB
364	nsmirnov	1.1	currDir = os.getcwd()
365			os.chdir(common.work_space.jobDir())
366	slacapra	1.9
367			tmpOrcarcList=[]
368			for PubDBversion in self.pubDBData.keys():
369			if len(self.pubDBData[PubDBversion])>0 :
370			#print (" PubDB-style : %s"%(PubDBversion))
371			if PubDBversion=='newPubDB' :
372	slacapra	1.19	self.builder = orcarcBuilder.orcarcBuilder(cfg_params)
373	slacapra	1.9	else :
374			self.builder = orcarcBuilderOld.orcarcBuilderOld()
375			tmpAllOrcarcs = self.builder.createOrcarcAndInit(self.pubDBData[PubDBversion])
376			tmpOrcarcList.append(tmpAllOrcarcs)
377			#print 'version ',PubDBversion,' tmpAllOrcarcs ', tmpAllOrcarcs
378
379			#print tmpOrcarcList
380	nsmirnov	1.1	os.chdir(currDir)
381
382			self.maxEvents=0
383	slacapra	1.9	for tmpAllOrcarcs in tmpOrcarcList:
384			for o in tmpAllOrcarcs:
385			numEvReq=self.total_number_of_events
386			if ((numEvReq == '-1') \| (numEvReq <= o.Nevents)):
387			self.allOrcarcs.append(o)
388	slacapra	1.31	if (int(o.Nevents) >= self.maxEvents):
389			self.maxEvents= int(o.Nevents)
390	slacapra	1.9	pass
391	nsmirnov	1.1	pass
392			pass
393
394			# set maximum number of event available
395
396			# I save to a file self.allOrcarcs
397
398			PubDBSummaryFile = open(common.work_space.shareDir()+'PubDBSummaryFile','w')
399			for o in self.allOrcarcs:
400			for d in o.content():
401			PubDBSummaryFile.write(d)
402			PubDBSummaryFile.write(' ')
403			pass
404			PubDBSummaryFile.write('\n')
405			pass
406			PubDBSummaryFile.close()
407	fanzago	1.29	### fede
408			for o in self.allOrcarcs:
409			o.dump()
410	nsmirnov	1.1	pass
411
412			# build a list of sites
413			ces= []
414			for o in self.allOrcarcs:
415			ces.append(o.CE)
416			pass
417
418			if len(ces)==0:
419	slacapra	1.9	msg = 'No PubDBs publish correct catalogs or enough events! '
420	nsmirnov	1.1	msg += `self.total_number_of_events`
421			raise CrabException(msg)
422
423	nsmirnov	1.3	common.logger.debug(6, "List of CEs: "+str(ces))
424	slacapra	1.6	common.analisys_common_info['sites']=ces
425	nsmirnov	1.1
426			return
427
428			def nJobs(self):
429			# TODO: should not be here !
430			# JobType should have no internal knowledge about submitted jobs
431			# One possibility is to use len(common.job_list).
432			""" return the number of job to be created """
433	slacapra	1.6	return len(common.job_list)
434	nsmirnov	1.5
435			def prepareSteeringCards(self):
436			"""
437			modify the orcarc card provided by the user,
438			writing a new card into share dir
439			"""
440			infile = ''
441			try:
442			infile = open(self.orcarc_file,'r')
443			except:
444			self.orcarc_file = 'empty.orcarc'
445			cmd='touch '+self.orcarc_file
446	slacapra	1.9	runCommand(cmd)
447	nsmirnov	1.5	infile = open(self.orcarc_file,'r')
448
449	slacapra	1.9	outfile = open(common.work_space.jobDir()+self.name()+'.orcarc', 'w')
450	nsmirnov	1.5
451			inline=infile.readlines()
452			### remove from user card these lines ###
453	slacapra	1.9	wordRemove=['InputFileCatalogURL', 'InputCollections', 'FirstEvent', 'MaxEvents', 'TFileAdaptor']
454			for line in inline:
455			word = string.strip(string.split(line,'=')[0])
456	slacapra	1.8
457	slacapra	1.9	if word not in wordRemove:
458			outfile.write(line)
459			else:
460			continue
461			pass
462
463			outfile.write('\n\n##### The following cards have been created by CRAB: DO NOT TOUCH #####\n')
464			outfile.write('TFileAdaptor = true\n')
465
466	slacapra	1.10	if (self.ML) :
467			outfile.write('MonalisaAddPid = false\n')
468			outfile.write('ExtraPackages=RecApplPlugins\n')
469			outfile.write('MonRecAlisaBuilder=true\n')
470			## TaskId is username+crab_0_date_time : that should be unique
471			TaskID = os.getlogin()+'_'+string.split(common.work_space.topDir(),'/')[-2]
472			outfile.write('MonalisaApplName='+TaskID+'\n')
473			outfile.write('MonalisaNode=192.91.245.5\n')
474			outfile.write('MonalisaPort=58884\n')
475			pass
476
477	slacapra	1.9	outfile.write('InputCollections=/System/'+self.owner+'/'+self.dataset+'/'+self.dataset+'\n')
478	slacapra	1.8
479	nsmirnov	1.5	infile.close()
480			outfile.close()
481			return
482	nsmirnov	1.1
483			def modifySteeringCards(self, nj):
484			"""
485			Creates steering cards file modifying a template file
486			"""
487			return
488
489			def cardsBaseName(self):
490			"""
491			Returns name of user orcarc card-file
492			"""
493			return os.path.split (self.orcarc_file)[1]
494
495	fanzago	1.27	### content of input_sanbdox ...
496	nsmirnov	1.1	def inputSandbox(self, nj):
497			"""
498			Returns a list of filenames to be put in JDL input sandbox.
499			"""
500			inp_box = []
501	corvo	1.13	# dict added to delete duplicate from input sandbox file list
502			seen = {}
503	slacapra	1.7	## code
504	fanzago	1.12	if os.path.isfile(self.tgzNameWithPath):
505			inp_box.append(self.tgzNameWithPath)
506	slacapra	1.7	## orcarc
507	nsmirnov	1.1	for o in self.allOrcarcs:
508			for f in o.fileList():
509	corvo	1.13	if (f not in seen.keys()):
510			inp_box.append(common.work_space.jobDir()+f)
511			seen[f] = 1
512	slacapra	1.9
513	slacapra	1.7	## config
514	nsmirnov	1.1	inp_box.append(common.job_list[nj].configFilename())
515	slacapra	1.7	## additional input files
516	fanzago	1.26	#inp_box = inp_box + self.additional_inbox_files
517			#print "sono inputSandbox, inp_box = ", inp_box
518	nsmirnov	1.1	return inp_box
519
520	fanzago	1.15	### and of output_sandbox
521	nsmirnov	1.1	def outputSandbox(self, nj):
522			"""
523			Returns a list of filenames to be put in JDL output sandbox.
524			"""
525			out_box = []
526
527	slacapra	1.9	stdout=common.job_list[nj].stdout()
528			stderr=common.job_list[nj].stderr()
529	fanzago	1.15	#out_box.append(stdout)
530			#out_box.append(stderr)
531	slacapra	1.9
532	slacapra	1.7	## User Declared output files
533	slacapra	1.9	for out in self.output_file:
534	fanzago	1.14	n_out = nj + 1
535	fanzago	1.25	#FEDE
536			#out_box.append(self.version+'/'+self.numberFile_(out,str(n_out)))
537			out_box.append(self.numberFile_(out,str(n_out)))
538	nsmirnov	1.1	return out_box
539	slacapra	1.7
540	slacapra	1.9	def numberFile_(self, file, txt):
541			"""
542			append _'txt' before last extension of a file
543			"""
544			p = string.split(file,".")
545			# take away last extension
546			name = p[0]
547			for x in p[1:-1]:
548			name=name+"."+x
549			# add "_txt"
550			if len(p)>1:
551			ext = p[len(p)-1]
552	fanzago	1.14	#result = name + '_' + str(txt) + "." + ext
553			result = name + '_' + txt + "." + ext
554	slacapra	1.9	else:
555	fanzago	1.14	#result = name + '_' + str(txt)
556			result = name + '_' + txt
557	slacapra	1.9
558			return result
559
560
561	slacapra	1.7	def stdOut(self):
562			return self.stdOut_
563
564			def stdErr(self):
565			return self.stdErr_