CRAB/python/Reporter.py

import os, common, string
from Actor import *
from crab_util import *
from ProdCommon.FwkJobRep.ReportParser import readJobReport
try: # Can remove when CMSSW 3.7 and earlier are dropped
    from FWCore.PythonUtilities.LumiList import LumiList
except ImportError:
    from LumiList import LumiList

try: # FUTURE: Python 2.6, prior to 2.6 requires simplejson
    import json
except:
    import simplejson as json


class Reporter(Actor):
    """ A class to report a short summary of the info of a task, including what
    is needed for user analysis, such as #events requestes/done, integrated
    lumi and so one.
    """
    def __init__(self, cfg_params):
        self.cfg_params = cfg_params
        self.fjrDirectory = cfg_params.get('USER.outputdir' ,
                                           common.work_space.resDir()) + '/'
        return

    def getInputRunLumi(self, file):
        import xml.dom.minidom

        dom = xml.dom.minidom.parse(file)
        ll=[]

        for elem in dom.getElementsByTagName("Job"):
            nJob = int(elem.getAttribute("JobID"))
            #print "---> nJob = ", nJob
            lumis          = elem.getAttribute('Lumis')
            #print "--->>> lumis = ", str(lumis)
            #lumis = '193752:1'
            #lumis = '193752:1-193752:5,193774:1-193774:5,193775:1'
            if lumis:
                tmp=str.split(str(lumis), ",")
                #print "tmp = ", tmp
            else:
                return
                

            #tmp = [193752:1-193752:5] [193774:1-193774:5]
            for entry in tmp:
                run_lumi=str.split(entry, "-")
                # run_lumi = [193752:1] [193752:5] 
                #print"run_lumi = ", run_lumi
                if len(run_lumi) == 0: pass
                if len(run_lumi) == 1:
                    lumi = str.split(run_lumi[0],":")[1]
                    run = str.split(run_lumi[0],":")[0]
                    ll.append((run,int(lumi)))
    
                if len(run_lumi) == 2:
                    lumi_max = str.split(run_lumi[1],":")[1]
                    lumi_min = str.split(run_lumi[0],":")[1]
                    #print "lumi_min = ", lumi_min
                    #print "lumi_max = ", lumi_max
                    run = str.split(run_lumi[1],":")[0]
                    #print "run = ", run
                    for count in range(int(lumi_min),int(lumi_max) + 1): 
                        ll.append((run,count))

        #print "alla fine ll = ", ll  

        if len(ll):
            lumiList = LumiList(lumis = ll)
            #print "lumiList = ", lumiList
            compactList = lumiList.getCompactList()
            #print "compactList = ", compactList

            totalLumiFilename = self.fjrDirectory + 'InputLumiSummaryOfTask.json'
            totalLumiSummary = open(totalLumiFilename, 'w')
            json.dump(compactList, totalLumiSummary)
            totalLumiSummary.write('\n')
            totalLumiSummary.close()
        return totalLumiFilename 

    def compareJsonFile(self,inputJsonFile):

        #if (self.fjrDirectory + 'lumiSummary.json'):
        reportFileName = self.fjrDirectory + 'lumiSummary.json'
        command = 'compareJSON.py --sub ' + inputJsonFile + ' ' + reportFileName + ' ' + self.fjrDirectory + 'missingLumiSummary.json'
        #common.logger.info(command)
        os.system(command)
        return

    def run(self):
        """
        The main method of the class: report status of a task
        """
        common.logger.debug( "Reporter::run() called")
        task = common._db.getTask()

        msg= "--------------------\n"
        msg +=  "Dataset: %s\n"%str(task['dataset'])
        if self.cfg_params.has_key('USER.copy_data') and int(self.cfg_params['USER.copy_data'])==1:
            msg+=  "Remote output :\n"
            ## TODO: SL should come from jobDB!
            from PhEDExDatasvcInfo import PhEDExDatasvcInfo

            stageout = PhEDExDatasvcInfo(self.cfg_params)
            endpoint, lfn, SE, SE_PATH, user = stageout.getEndpoint()
            #print endpoint, lfn, SE, SE_PATH, user

            msg+=  "SE: %s %s  srmPath: %s\n"%(self.cfg_params['USER.storage_element'],SE,endpoint)

        else:
            msg += "Local output: %s\n" % task['outputDirectory']
        #print task
        possible_status = [ 'Created',
                            'Undefined',
                            'Submitting',
                            'Submitted',
                            'NotSubmitted',
                            'Waiting',
                            'Ready',
                            'Scheduled',
                            'Running',
                            'Done',
                            'Killing',
                            'Killed',
                            'Aborted',
                            'Unknown',
                            'Done (Failed)',
                            'Cleared',
                            'Retrieved'
                            ]
        eventsRead=0
        eventsRequired=0
        filesRead=0
        filesRequired=0
        lumis = []
        for job in task.getJobs():
            if (job.runningJob['applicationReturnCode']!=0 or job.runningJob['wrapperReturnCode']!=0): continue
            # get FJR filename
            fjr = self.fjrDirectory + job['outputFiles'][-1]

            jobReport = readJobReport(fjr)
            if len(jobReport) > 0:
                inputFiles = jobReport[0].inputFiles
                for inputFile in inputFiles:
                    # Accumulate the list of lum sections run over
                    for run in inputFile.runs.keys():
                        for lumi in inputFile.runs[run]:
                            lumis.append((run, lumi))
                    filesRead+=1
                    eventsRead+=int(inputFile['EventsRead'])
                #print jobReport[0].inputFiles,'\n'
            else:
                pass
                #print 'no FJR avaialble for job #%s'%job['jobId']
            #print "--------------------------"

        # Compact and write the list of successful lumis

        lumiList = LumiList(lumis = lumis)
        compactList = lumiList.getCompactList()

        lumiFilename = task['outputDirectory'] + 'lumiSummary.json'
        lumiSummary = open(lumiFilename, 'w')
        json.dump(compactList, lumiSummary)
        lumiSummary.write('\n')
        lumiSummary.close()

        msg += "Total Events read: %s\n" % eventsRead
        msg += "Total Files read: %s\n" % filesRead
        msg += "Total Jobs : %s\n" % len(task.getJobs())
        msg += "Luminosity section summary file: %s\n" % lumiFilename
        list_ID={}

        # TEMPORARY by Fabio, to be removed
        # avoid clashes between glite_slc5 and glite schedulers when a server is used
        # otherwise, -report with a server requires a local scheduler
        if self.cfg_params.get('CRAB.server_name', None) is None:
            common.logger.debug( "Reporter updating task status")
            task = common.scheduler.queryEverything(task['id'])

        for st in possible_status:
            list_ID = common._db.queryAttrRunJob({'statusScheduler':st},'jobId')
            if (len(list_ID)>0):
                msg+=  "   # Jobs: %s:%s\n"%(str(st),len(list_ID))
            pass
        msg+=  "\n----------------------------\n"
        common.logger.info(msg)


        file = common.work_space.shareDir() + 'arguments.xml'
        #print "file = ", file
        
        ### starting from the arguments.xml file, a json file containing the run:lumi
        ### that should be analyzed with the task
        inputRunLumiFileName = self.getInputRunLumi(file)

        
        ### missing lumi to analyze: starting from lumimask or from argument file
        ### calculate the difference with report.json
        ### if a lumimask is used in the crab.cfg
        if (self.cfg_params.get('CMSSW.lumi_mask')): 
            lumimask=self.cfg_params.get('CMSSW.lumi_mask')
            #print "lumimask = ", lumimask 
            self.compareJsonFile(lumimask)
        ### without lumimask    
        elif (inputRunLumiFileName):
            self.compareJsonFile(inputRunLumiFileName)
        else:
            common.logger.info("no json file to compare")
        return

Revision:	1.19
Committed:	Wed Sep 5 16:02:43 2012 UTC (12 years, 7 months ago) by fanzago
Content type:	text/x-python
Branch:	MAIN
Changes since 1.18:	+87 -1 lines
Log Message:	added the missing_lumi.json file in the report, savannah bug 95967
#	User	Rev	Content
1	slacapra	1.1	import os, common, string
2			from Actor import *
3			from crab_util import *
4	ewv	1.13	from ProdCommon.FwkJobRep.ReportParser import readJobReport
5	ewv	1.17	try: # Can remove when CMSSW 3.7 and earlier are dropped
6			from FWCore.PythonUtilities.LumiList import LumiList
7			except ImportError:
8			from LumiList import LumiList
9	ewv	1.13
10			try: # FUTURE: Python 2.6, prior to 2.6 requires simplejson
11			import json
12			except:
13			import simplejson as json
14
15	slacapra	1.1
16			class Reporter(Actor):
17			""" A class to report a short summary of the info of a task, including what
18			is needed for user analysis, such as #events requestes/done, integrated
19			lumi and so one.
20			"""
21			def __init__(self, cfg_params):
22			self.cfg_params = cfg_params
23	ewv	1.16	self.fjrDirectory = cfg_params.get('USER.outputdir' ,
24			common.work_space.resDir()) + '/'
25	slacapra	1.1	return
26
27	fanzago	1.19	def getInputRunLumi(self, file):
28			import xml.dom.minidom
29
30			dom = xml.dom.minidom.parse(file)
31			ll=[]
32
33			for elem in dom.getElementsByTagName("Job"):
34			nJob = int(elem.getAttribute("JobID"))
35			#print "---> nJob = ", nJob
36			lumis = elem.getAttribute('Lumis')
37			#print "--->>> lumis = ", str(lumis)
38			#lumis = '193752:1'
39			#lumis = '193752:1-193752:5,193774:1-193774:5,193775:1'
40			if lumis:
41			tmp=str.split(str(lumis), ",")
42			#print "tmp = ", tmp
43			else:
44			return
45
46
47			#tmp = [193752:1-193752:5] [193774:1-193774:5]
48			for entry in tmp:
49			run_lumi=str.split(entry, "-")
50			# run_lumi = [193752:1] [193752:5]
51			#print"run_lumi = ", run_lumi
52			if len(run_lumi) == 0: pass
53			if len(run_lumi) == 1:
54			lumi = str.split(run_lumi[0],":")[1]
55			run = str.split(run_lumi[0],":")[0]
56			ll.append((run,int(lumi)))
57
58			if len(run_lumi) == 2:
59			lumi_max = str.split(run_lumi[1],":")[1]
60			lumi_min = str.split(run_lumi[0],":")[1]
61			#print "lumi_min = ", lumi_min
62			#print "lumi_max = ", lumi_max
63			run = str.split(run_lumi[1],":")[0]
64			#print "run = ", run
65			for count in range(int(lumi_min),int(lumi_max) + 1):
66			ll.append((run,count))
67
68			#print "alla fine ll = ", ll
69
70			if len(ll):
71			lumiList = LumiList(lumis = ll)
72			#print "lumiList = ", lumiList
73			compactList = lumiList.getCompactList()
74			#print "compactList = ", compactList
75
76			totalLumiFilename = self.fjrDirectory + 'InputLumiSummaryOfTask.json'
77			totalLumiSummary = open(totalLumiFilename, 'w')
78			json.dump(compactList, totalLumiSummary)
79			totalLumiSummary.write('\n')
80			totalLumiSummary.close()
81			return totalLumiFilename
82
83			def compareJsonFile(self,inputJsonFile):
84
85			#if (self.fjrDirectory + 'lumiSummary.json'):
86			reportFileName = self.fjrDirectory + 'lumiSummary.json'
87			command = 'compareJSON.py --sub ' + inputJsonFile + ' ' + reportFileName + ' ' + self.fjrDirectory + 'missingLumiSummary.json'
88			#common.logger.info(command)
89			os.system(command)
90			return
91
92	slacapra	1.1	def run(self):
93			"""
94			The main method of the class: report status of a task
95			"""
96	spiga	1.4	common.logger.debug( "Reporter::run() called")
97	spiga	1.7	task = common._db.getTask()
98
99	spiga	1.6	msg= "--------------------\n"
100			msg += "Dataset: %s\n"%str(task['dataset'])
101	slacapra	1.2	if self.cfg_params.has_key('USER.copy_data') and int(self.cfg_params['USER.copy_data'])==1:
102	spiga	1.6	msg+= "Remote output :\n"
103	slacapra	1.2	## TODO: SL should come from jobDB!
104			from PhEDExDatasvcInfo import PhEDExDatasvcInfo
105	spiga	1.7
106	slacapra	1.2	stageout = PhEDExDatasvcInfo(self.cfg_params)
107			endpoint, lfn, SE, SE_PATH, user = stageout.getEndpoint()
108			#print endpoint, lfn, SE, SE_PATH, user
109
110	spiga	1.6	msg+= "SE: %s %s srmPath: %s\n"%(self.cfg_params['USER.storage_element'],SE,endpoint)
111	ewv	1.13
112	slacapra	1.2	else:
113	ewv	1.16	msg += "Local output: %s\n" % task['outputDirectory']
114	slacapra	1.2	#print task
115	slacapra	1.1	possible_status = [ 'Created',
116			'Undefined',
117			'Submitting',
118			'Submitted',
119	slacapra	1.3	'NotSubmitted',
120	slacapra	1.1	'Waiting',
121			'Ready',
122			'Scheduled',
123			'Running',
124			'Done',
125			'Killing',
126			'Killed',
127			'Aborted',
128			'Unknown',
129			'Done (Failed)',
130			'Cleared',
131	spiga	1.5	'Retrieved'
132	slacapra	1.1	]
133			eventsRead=0
134	slacapra	1.2	eventsRequired=0
135	slacapra	1.1	filesRead=0
136	slacapra	1.2	filesRequired=0
137	spiga	1.15	lumis = []
138	slacapra	1.1	for job in task.getJobs():
139	fanzago	1.18	if (job.runningJob['applicationReturnCode']!=0 or job.runningJob['wrapperReturnCode']!=0): continue
140	slacapra	1.1	# get FJR filename
141	ewv	1.16	fjr = self.fjrDirectory + job['outputFiles'][-1]
142
143	slacapra	1.1	jobReport = readJobReport(fjr)
144	ewv	1.13	if len(jobReport) > 0:
145			inputFiles = jobReport[0].inputFiles
146	slacapra	1.1	for inputFile in inputFiles:
147	ewv	1.13	# Accumulate the list of lum sections run over
148			for run in inputFile.runs.keys():
149			for lumi in inputFile.runs[run]:
150	spiga	1.15	lumis.append((run, lumi))
151	slacapra	1.1	filesRead+=1
152			eventsRead+=int(inputFile['EventsRead'])
153			#print jobReport[0].inputFiles,'\n'
154			else:
155	slacapra	1.3	pass
156			#print 'no FJR avaialble for job #%s'%job['jobId']
157	slacapra	1.1	#print "--------------------------"
158	ewv	1.13
159	spiga	1.15	# Compact and write the list of successful lumis
160	ewv	1.13
161	spiga	1.15	lumiList = LumiList(lumis = lumis)
162			compactList = lumiList.getCompactList()
163	ewv	1.13
164			lumiFilename = task['outputDirectory'] + 'lumiSummary.json'
165			lumiSummary = open(lumiFilename, 'w')
166			json.dump(compactList, lumiSummary)
167			lumiSummary.write('\n')
168			lumiSummary.close()
169
170			msg += "Total Events read: %s\n" % eventsRead
171			msg += "Total Files read: %s\n" % filesRead
172			msg += "Total Jobs : %s\n" % len(task.getJobs())
173			msg += "Luminosity section summary file: %s\n" % lumiFilename
174	slacapra	1.1	list_ID={}
175	farinafa	1.14
176	ewv	1.16	# TEMPORARY by Fabio, to be removed
177	farinafa	1.14	# avoid clashes between glite_slc5 and glite schedulers when a server is used
178	ewv	1.16	# otherwise, -report with a server requires a local scheduler
179	farinafa	1.14	if self.cfg_params.get('CRAB.server_name', None) is None:
180			common.logger.debug( "Reporter updating task status")
181			task = common.scheduler.queryEverything(task['id'])
182
183	slacapra	1.1	for st in possible_status:
184			list_ID = common._db.queryAttrRunJob({'statusScheduler':st},'jobId')
185			if (len(list_ID)>0):
186	spiga	1.6	msg+= " # Jobs: %s:%s\n"%(str(st),len(list_ID))
187	slacapra	1.1	pass
188	spiga	1.6	msg+= "\n----------------------------\n"
189	ewv	1.13	common.logger.info(msg)
190	fanzago	1.19
191
192			file = common.work_space.shareDir() + 'arguments.xml'
193			#print "file = ", file
194
195			### starting from the arguments.xml file, a json file containing the run:lumi
196			### that should be analyzed with the task
197			inputRunLumiFileName = self.getInputRunLumi(file)
198
199
200			### missing lumi to analyze: starting from lumimask or from argument file
201			### calculate the difference with report.json
202			### if a lumimask is used in the crab.cfg
203			if (self.cfg_params.get('CMSSW.lumi_mask')):
204			lumimask=self.cfg_params.get('CMSSW.lumi_mask')
205			#print "lumimask = ", lumimask
206			self.compareJsonFile(lumimask)
207			### without lumimask
208			elif (inputRunLumiFileName):
209			self.compareJsonFile(inputRunLumiFileName)
210			else:
211			common.logger.info("no json file to compare")
212	ewv	1.13	return
213