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.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:
            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)
        except:
            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

        ## additional input files
        self.additional_inbox_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)

        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_()

        # 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 'Error: Too few arguments' +$narg+ \n"
        txt += "    exit 1\n"
        txt += "fi\n"
        txt += "NJob = $1\n"
        txt += "FirstEvent = $2\n"
        txt += "MaxEvent = $3\n"
        txt += "\n"

        # Prepare JobType-specific part
        scram = self.scram.commandName()
        txt += '\n'
        txt += scram+' project ORCA '+self.version+'\n'
        txt += 'status=$?\n'
        txt += 'if [ $status != 0 ] ; then\n'
        txt += '   echo "Warning: ORCA '+self.version+' not found on `hostname`" \n'
        txt += '   exit 1 \n'
        txt += 'fi \n'
        txt += 'cd '+self.version+'\n'
        txt += 'eval `'+scram+' runtime -sh`\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 += '  cp $RUNTIME_AREA/init_$CE.sh init.sh\n'
        txt += 'fi\n'
        txt += 'echo "***** cat .orcarc *********"\n'
        txt += 'cat .orcarc\n'
        txt += 'echo "****** end .orcarc ********"\n'
        txt += '\n'
        txt += 'chmod +x ./init.sh\n'
        txt += './init.sh\n'
        txt += 'exitStatus=$?\n'
        txt += 'if [ $exitStatus != 0 ] ; then\n'
        txt += '  echo "StageIn init script failed!"\n'
        txt += '  exit $exitStatus\n'
        txt += 'fi\n'

        txt += 'echo FirstEvent = ${FirstEvent} >> .orcarc\n'
        txt += 'echo MaxEvent = ${MaxEvent} >> .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.tgz):
            txt += 'echo "tar xzvf ../'+os.path.basename(self.tgz)+'"\n'
            txt += 'tar xzvf ../'+os.path.basename(self.tgz)+'\n'
            txt += 'untar_status=$? \n'
            txt += 'if [ $untar_status -ne 0 ]; then \n'
            txt += '   echo "Untarring .tgz file failed ... exiting" \n'
            txt += '   exit 1 \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'
        for fileWithSuffix in self.output_file:
            output_file_num = self.numberFile_(fileWithSuffix, '$NJob')
            txt += 'cp '+fileWithSuffix+' '+output_file_num+'\n'
            pass
        return txt

    def executableName(self):
        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()
                    else :
                        self.builder = orcarcBuilderOld.orcarcBuilderOld()
                    tmpAllOrcarcs = self.builder.createOrcarcAndInit(self.pubDBData[PubDBversion])
                    #print "tmpAllOrcarcs", tmpAllOrcarcs  
                    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 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(orcarcBuilder.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.RefDBError:
    #             msg = 'ERROR ***: accessing PubDB'
    #             raise CrabException(msg)

    #         self.CollIDs = self.pubdb.findAllCollections()
    #         self.SelectPubDBURLs=self.pubdb.findPubDBs(self.CollIDs)
    #         self.pubDBResults = self.pubdb.getAllPubDBData(self.CollIDs,self.SelectPubDBURLs)
    #         #self.pubDBResults = self.pubdb.getAllPubDBsInfo()

    #         if len(self.pubDBResults)==0:
    #             msg = 'Owner Dataset not published with asked dataTiers! '+self.owner+' '+ self.dataset+' '+str(self.dataTiers)
    #             raise CrabException(msg)

    #         common.logger.debug(6, fun+": PubDB info ("+`len(self.pubDBResults)`+"):\/")
    #         for aa in self.pubDBResults:
    #             for bb in aa:
    #                 common.logger.debug(6, str(bb))
    #                 pass
    #             pass
    #         common.logger.debug(6, fun+": End of PubDB info\n")

    #         self.builder = orcarcBuilder.orcarcBuilder()

    #         currDir = os.getcwd()
    #         os.chdir(common.work_space.jobDir())
    #         tmpAllOrcarcs = self.builder.createOrcarcAndInit(self.pubDBResults)
    #         os.chdir(currDir)

    #         self.maxEvents=0
    #         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 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)
        #return int((self.total_number_of_events-1)/self.job_number_of_events)+1

    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')
        ## TODO put now all cards which are common to all jobs 
        ##   (like TFileAdaptor or Monalisa-related stuff)
        outfile.write('TFileAdaptor = true\n')

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

        infile.close()
        outfile.close()
        return
    
    def modifySteeringCards(self, nj):
        # add jobs information to the orcarc card, 
        # starting from card into share dir 
        """
        Creates steering cards file modifying a template file
        """
        # infile =  open(common.work_space.shareDir()+self.cardsBaseName(), 'r')
        # outfile = open(common.job_list[nj].configFilename(),'w')  
        # ### job splitting      
        # firstEvent = common.jobDB.firstEvent(nj)
        # maxEvents = common.jobDB.maxEvents(nj)
        # #Nev_job = self.job_number_of_events
        # outfile.write('InputCollections=/System/'+self.owner+'/'+self.dataset+'/'+self.dataset+'\n')
        # outfile.write('FirstEvent = '+ str(firstEvent) +'\n')
        # 
        # outfile.write('MaxEvents = '+str(maxEvents)+'\n')

        # outfile.write(infile.read())
        # outfile.close()
        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 = []
        ## code
        self.tgz = self.tgzNameWithPath
        if os.path.isfile(self.tgz):
            inp_box.append(self.tgz)
        ## orcarc
        for o in self.allOrcarcs:
          for f in o.fileList():
            inp_box.append(common.work_space.jobDir()+f)

        ## config
        inp_box.append(common.job_list[nj].configFilename())
        ## additional input files
        inp_box = inp_box + self.additional_inbox_files
        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:
            out_box.append(self.version+'/'+self.numberFile_(out,str(nj)))
        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
        else:
          result = name + '_' + str(txt)
        
        return result


    def stdOut(self):
        return self.stdOut_

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