CMS 3D CMS Logo
MatrixInjector.py
Go to the documentation of this file.
1 from __future__ import print_function
2 import sys
3 import json
4 import os
5 import copy
6 import multiprocessing
7 import time
8 import re
9 
10 MAXWORKFLOWLENGTH = 81
11 
12 def performInjectionOptionTest(opt):
13  if opt.show:
14  print('Not injecting to wmagent in --show mode. Need to run the worklfows.')
15  sys.exit(-1)
16  if opt.wmcontrol=='init':
17  #init means it'll be in test mode
18  opt.nProcs=0
19  if opt.wmcontrol=='test':
20  #means the wf were created already, and we just dryRun it.
21  opt.dryRun=True
22  if opt.wmcontrol=='submit' and opt.nProcs==0:
23  print('Not injecting to wmagent in -j 0 mode. Need to run the worklfows.')
24  sys.exit(-1)
25  if opt.wmcontrol=='force':
26  print("This is an expert setting, you'd better know what you're doing")
27  opt.dryRun=True
28 
29 def upload_to_couch_oneArg(arguments):
30  from modules.wma import upload_to_couch
31  (filePath,labelInCouch,user,group,where) = arguments
32  cacheId=upload_to_couch(filePath,
33  labelInCouch,
34  user,
35  group,
36  test_mode=False,
37  url=where)
38  return cacheId
39 
40 
41 class MatrixInjector(object):
42 
43  def __init__(self,opt,mode='init',options=''):
44  self.count=1040
45 
46  self.dqmgui=None
47  self.wmagent=None
48  for k in options.split(','):
49  if k.startswith('dqm:'):
50  self.dqmgui=k.split(':',1)[-1]
51  elif k.startswith('wma:'):
52  self.wmagent=k.split(':',1)[-1]
53 
54  self.testMode=((mode!='submit') and (mode!='force'))
55  self.version =1
56  self.keep = opt.keep
57  self.memoryOffset = opt.memoryOffset
58  self.memPerCore = opt.memPerCore
59  self.numberEventsInLuminosityBlock = opt.numberEventsInLuminosityBlock
60  self.batchName = ''
61  self.batchTime = str(int(time.time()))
62  if(opt.batchName):
63  self.batchName = '__'+opt.batchName+'-'+self.batchTime
64 
65  #wagemt stuff
66  if not self.wmagent:
67  self.wmagent=os.getenv('WMAGENT_REQMGR')
68  if not self.wmagent:
69  if not opt.testbed :
70  self.wmagent = 'cmsweb.cern.ch'
71  self.DbsUrl = "https://"+self.wmagent+"/dbs/prod/global/DBSReader"
72  else :
73  self.wmagent = 'cmsweb-testbed.cern.ch'
74  self.DbsUrl = "https://"+self.wmagent+"/dbs/int/global/DBSReader"
75 
76  if not self.dqmgui:
77  self.dqmgui="https://cmsweb.cern.ch/dqm/relval"
78  #couch stuff
79  self.couch = 'https://'+self.wmagent+'/couchdb'
80 # self.couchDB = 'reqmgr_config_cache'
81  self.couchCache={} # so that we do not upload like crazy, and recyle cfgs
82  self.user = os.getenv('USER')
83  self.group = 'ppd'
84  self.label = 'RelValSet_'+os.getenv('CMSSW_VERSION').replace('-','')+'_v'+str(self.version)
85  self.speciallabel=''
86  if opt.label:
87  self.speciallabel= '_'+opt.label
88  self.longWFName = []
89 
90  if not os.getenv('WMCORE_ROOT'):
91  print('\n\twmclient is not setup properly. Will not be able to upload or submit requests.\n')
92  if not self.testMode:
93  print('\n\t QUIT\n')
94  sys.exit(-18)
95  else:
96  print('\n\tFound wmclient\n')
97 
98  self.defaultChain={
99  "RequestType" : "TaskChain", #this is how we handle relvals
100  "SubRequestType" : "RelVal", #this is how we handle relvals, now that TaskChain is also used for central MC production
101  "RequestPriority": 500000,
102  "Requestor": self.user, #Person responsible
103  "Group": self.group, #group for the request
104  "CMSSWVersion": os.getenv('CMSSW_VERSION'), #CMSSW Version (used for all tasks in chain)
105  "Campaign": os.getenv('CMSSW_VERSION'), # = AcquisitionEra, will be reset later to the one of first task, will both be the CMSSW_VERSION
106  "ScramArch": os.getenv('SCRAM_ARCH'), #Scram Arch (used for all tasks in chain)
107  "ProcessingVersion": self.version, #Processing Version (used for all tasks in chain)
108  "GlobalTag": None, #Global Tag (overridden per task)
109  "ConfigCacheUrl": self.couch, #URL of CouchDB containing Config Cache
110  "DbsUrl": self.DbsUrl,
111  #- Will contain all configs for all Tasks
112  #"SiteWhitelist" : ["T2_CH_CERN", "T1_US_FNAL"], #Site whitelist
113  "TaskChain" : None, #Define number of tasks in chain.
114  "nowmTasklist" : [], #a list of tasks as we put them in
115  "Multicore" : 1, # do not set multicore for the whole chain
116  "Memory" : 3000,
117  "SizePerEvent" : 1234,
118  "TimePerEvent" : 10,
119  "PrepID": os.getenv('CMSSW_VERSION')
120  }
121 
122  self.defaultHarvest={
123  "EnableHarvesting" : "True",
124  "DQMUploadUrl" : self.dqmgui,
125  "DQMConfigCacheID" : None,
126  "Multicore" : 1 # hardcode Multicore to be 1 for Harvest
127  }
128 
129  self.defaultScratch={
130  "TaskName" : None, #Task Name
131  "ConfigCacheID" : None, #Generator Config id
132  "GlobalTag": None,
133  "SplittingAlgo" : "EventBased", #Splitting Algorithm
134  "EventsPerJob" : None, #Size of jobs in terms of splitting algorithm
135  "EventsPerLumi" : None,
136  "RequestNumEvents" : None, #Total number of events to generate
137  "Seeding" : "AutomaticSeeding", #Random seeding method
138  "PrimaryDataset" : None, #Primary Dataset to be created
139  "nowmIO": {},
140  "Multicore" : opt.nThreads, # this is the per-taskchain Multicore; it's the default assigned to a task if it has no value specified
141  "KeepOutput" : False
142  }
143  self.defaultInput={
144  "TaskName" : "DigiHLT", #Task Name
145  "ConfigCacheID" : None, #Processing Config id
146  "GlobalTag": None,
147  "InputDataset" : None, #Input Dataset to be processed
148  "SplittingAlgo" : "LumiBased", #Splitting Algorithm
149  "LumisPerJob" : 10, #Size of jobs in terms of splitting algorithm
150  "nowmIO": {},
151  "Multicore" : opt.nThreads, # this is the per-taskchain Multicore; it's the default assigned to a task if it has no value specified
152  "KeepOutput" : False
153  }
154  self.defaultTask={
155  "TaskName" : None, #Task Name
156  "InputTask" : None, #Input Task Name (Task Name field of a previous Task entry)
157  "InputFromOutputModule" : None, #OutputModule name in the input task that will provide files to process
158  "ConfigCacheID" : None, #Processing Config id
159  "GlobalTag": None,
160  "SplittingAlgo" : "LumiBased", #Splitting Algorithm
161  "LumisPerJob" : 10, #Size of jobs in terms of splitting algorithm
162  "nowmIO": {},
163  "Multicore" : opt.nThreads, # this is the per-taskchain Multicore; it's the default assigned to a task if it has no value specified
164  "KeepOutput" : False
165  }
166 
167  self.chainDicts={}
168 
169 
170  @staticmethod
171  def get_wmsplit():
172  """
173  Return a "wmsplit" dictionary that contain non-default LumisPerJob values
174  """
175  wmsplit = {}
176  try:
177  #from Configuration.PyReleaseValidation.relval_steps import wmsplit
178  wmsplit = {}
179  wmsplit['DIGIHI']=5
180  wmsplit['RECOHI']=5
181  wmsplit['HLTD']=5
182  wmsplit['RECODreHLT']=2
183  wmsplit['DIGIPU']=4
184  wmsplit['DIGIPU1']=4
185  wmsplit['RECOPU1']=1
186  wmsplit['DIGIUP15_PU50']=1
187  wmsplit['RECOUP15_PU50']=1
188  wmsplit['DIGIUP15_PU25']=1
189  wmsplit['RECOUP15_PU25']=1
190  wmsplit['DIGIUP15_PU25HS']=1
191  wmsplit['RECOUP15_PU25HS']=1
192  wmsplit['DIGIHIMIX']=5
193  wmsplit['RECOHIMIX']=5
194  wmsplit['RECODSplit']=1
195  wmsplit['SingleMuPt10_UP15_ID']=1
196  wmsplit['DIGIUP15_ID']=1
197  wmsplit['RECOUP15_ID']=1
198  wmsplit['TTbar_13_ID']=1
199  wmsplit['SingleMuPt10FS_ID']=1
200  wmsplit['TTbarFS_ID']=1
201  wmsplit['RECODR2_50nsreHLT']=5
202  wmsplit['RECODR2_25nsreHLT']=5
203  wmsplit['RECODR2_2016reHLT']=5
204  wmsplit['RECODR2_50nsreHLT_HIPM']=5
205  wmsplit['RECODR2_25nsreHLT_HIPM']=5
206  wmsplit['RECODR2_2016reHLT_HIPM']=1
207  wmsplit['RECODR2_2016reHLT_skimSingleMu']=1
208  wmsplit['RECODR2_2016reHLT_skimDoubleEG']=1
209  wmsplit['RECODR2_2016reHLT_skimMuonEG']=1
210  wmsplit['RECODR2_2016reHLT_skimJetHT']=1
211  wmsplit['RECODR2_2016reHLT_skimMET']=1
212  wmsplit['RECODR2_2016reHLT_skimSinglePh']=1
213  wmsplit['RECODR2_2016reHLT_skimMuOnia']=1
214  wmsplit['RECODR2_2016reHLT_skimSingleMu_HIPM']=1
215  wmsplit['RECODR2_2016reHLT_skimDoubleEG_HIPM']=1
216  wmsplit['RECODR2_2016reHLT_skimMuonEG_HIPM']=1
217  wmsplit['RECODR2_2016reHLT_skimJetHT_HIPM']=1
218  wmsplit['RECODR2_2016reHLT_skimMET_HIPM']=1
219  wmsplit['RECODR2_2016reHLT_skimSinglePh_HIPM']=1
220  wmsplit['RECODR2_2016reHLT_skimMuOnia_HIPM']=1
221  wmsplit['RECODR2_2017reHLT_Prompt']=1
222  wmsplit['RECODR2_2017reHLT_skimSingleMu_Prompt_Lumi']=1
223  wmsplit['RECODR2_2017reHLT_skimDoubleEG_Prompt']=1
224  wmsplit['RECODR2_2017reHLT_skimMET_Prompt']=1
225  wmsplit['RECODR2_2017reHLT_skimMuOnia_Prompt']=1
226  wmsplit['RECODR2_2017reHLT_Prompt_L1TEgDQM']=1
227  wmsplit['RECODR2_2018reHLT_Prompt']=1
228  wmsplit['RECODR2_2018reHLT_skimSingleMu_Prompt_Lumi']=1
229  wmsplit['RECODR2_2018reHLT_skimDoubleEG_Prompt']=1
230  wmsplit['RECODR2_2018reHLT_skimJetHT_Prompt']=1
231  wmsplit['RECODR2_2018reHLT_skimMET_Prompt']=1
232  wmsplit['RECODR2_2018reHLT_skimMuOnia_Prompt']=1
233  wmsplit['RECODR2_2018reHLT_skimEGamma_Prompt_L1TEgDQM']=1
234  wmsplit['RECODR2_2018reHLT_skimMuonEG_Prompt']=1
235  wmsplit['RECODR2_2018reHLT_skimCharmonium_Prompt']=1
236  wmsplit['RECODR2_2018reHLT_skimJetHT_Prompt_HEfail']=1
237  wmsplit['RECODR2_2018reHLT_skimJetHT_Prompt_BadHcalMitig']=1
238  wmsplit['RECODR2_2018reHLTAlCaTkCosmics_Prompt']=1
239  wmsplit['RECODR2_2018reHLT_skimDisplacedJet_Prompt']=1
240  wmsplit['RECODR2_2018reHLT_ZBPrompt']=1
241  wmsplit['RECODR2_2018reHLT_Offline']=1
242  wmsplit['RECODR2_2018reHLT_skimSingleMu_Offline_Lumi']=1
243  wmsplit['RECODR2_2018reHLT_skimDoubleEG_Offline']=1
244  wmsplit['RECODR2_2018reHLT_skimJetHT_Offline']=1
245  wmsplit['RECODR2_2018reHLT_skimMET_Offline']=1
246  wmsplit['RECODR2_2018reHLT_skimMuOnia_Offline']=1
247  wmsplit['RECODR2_2018reHLT_skimEGamma_Offline_L1TEgDQM']=1
248  wmsplit['RECODR2_2018reHLT_skimMuonEG_Offline']=1
249  wmsplit['RECODR2_2018reHLT_skimCharmonium_Offline']=1
250  wmsplit['RECODR2_2018reHLT_skimJetHT_Offline_HEfail']=1
251  wmsplit['RECODR2_2018reHLT_skimJetHT_Offline_BadHcalMitig']=1
252  wmsplit['RECODR2_2018reHLTAlCaTkCosmics_Offline']=1
253  wmsplit['RECODR2_2018reHLT_skimDisplacedJet_Offline']=1
254  wmsplit['RECODR2_2018reHLT_ZBOffline']=1
255  wmsplit['HLTDR2_50ns']=1
256  wmsplit['HLTDR2_25ns']=1
257  wmsplit['HLTDR2_2016']=1
258  wmsplit['HLTDR2_2017']=1
259  wmsplit['HLTDR2_2018']=1
260  wmsplit['HLTDR2_2018_BadHcalMitig']=1
261  wmsplit['Hadronizer']=1
262  wmsplit['DIGIUP15']=1
263  wmsplit['RECOUP15']=1
264  wmsplit['RECOAODUP15']=5
265  wmsplit['DBLMINIAODMCUP15NODQM']=5
266  wmsplit['DigiFull']=5
267  wmsplit['RecoFull']=5
268  wmsplit['DigiFullPU']=1
269  wmsplit['RecoFullPU']=1
270  wmsplit['RECOHID11']=1
271  wmsplit['DigiFullTriggerPU_2026D17PU'] = 1
272  wmsplit['RecoFullGlobalPU_2026D17PU']=1
273  wmsplit['DIGIUP17']=1
274  wmsplit['RECOUP17']=1
275  wmsplit['DIGIUP17_PU25']=1
276  wmsplit['RECOUP17_PU25']=1
277  wmsplit['DIGICOS_UP16']=1
278  wmsplit['RECOCOS_UP16']=1
279  wmsplit['DIGICOS_UP17']=1
280  wmsplit['RECOCOS_UP17']=1
281  wmsplit['DIGICOS_UP18']=1
282  wmsplit['RECOCOS_UP18']=1
283  wmsplit['DIGICOS_UP21']=1
284  wmsplit['RECOCOS_UP21']=1
285  wmsplit['HYBRIDRepackHI2015VR']=1
286  wmsplit['HYBRIDZSHI2015']=1
287  wmsplit['RECOHID15']=1
288  wmsplit['RECOHID18']=1
289  wmsplit['DigiFullTriggerPU_2026D35PU'] = 1
290  wmsplit['RecoFullGlobalPU_2026D35PU']=1
291  wmsplit['DigiFullTriggerPU_2026D41PU'] = 1
292  wmsplit['RecoFullGlobalPU_2026D41PU']=1
293  wmsplit['DigiFullTriggerPU_2026D43PU'] = 1
294  wmsplit['RecoFullGlobalPU_2026D43PU']=1
295  wmsplit['DigiFullTriggerPU_2026D44PU'] = 1
296  wmsplit['RecoFullGlobalPU_2026D44PU']=1
297  wmsplit['DigiFullTriggerPU_2026D45PU'] = 1
298  wmsplit['RecoFullGlobalPU_2026D45PU']=1
299  wmsplit['DigiFullTriggerPU_2026D46PU'] = 1
300  wmsplit['RecoFullGlobalPU_2026D46PU']=1
301  wmsplit['DigiFullTriggerPU_2026D47PU'] = 1
302  wmsplit['RecoFullGlobalPU_2026D47PU']=1
303  wmsplit['DigiFullTriggerPU_2026D48PU'] = 1
304  wmsplit['RecoFullGlobalPU_2026D48PU']=1
305  wmsplit['DigiFullTriggerPU_2026D49PU'] = 1
306  wmsplit['RecoFullGlobalPU_2026D49PU']=1
307  wmsplit['DigiFullTriggerPU_2026D50PU'] = 1
308  wmsplit['RecoFullGlobalPU_2026D50PU']=1
309  wmsplit['DigiFullTriggerPU_2026D51PU'] = 1
310  wmsplit['RecoFullGlobalPU_2026D51PU']=1
311  wmsplit['DigiFullTriggerPU_2026D52PU'] = 1
312  wmsplit['RecoFullGlobalPU_2026D52PU']=1
313  wmsplit['DigiFullTriggerPU_2026D57PU'] = 1
314  wmsplit['RecoFullGlobalPU_2026D57PU']=1
315  wmsplit['DigiFullTriggerPU_2026D58PU'] = 1
316  wmsplit['RecoFullGlobalPU_2026D58PU']=1
317  except Exception as ex:
318  print('Exception while building a wmsplit dictionary: %s' % (str(ex)))
319  return {}
320 
321  return wmsplit
322 
323  def prepare(self,mReader, directories, mode='init'):
324  wmsplit = MatrixInjector.get_wmsplit()
325  acqEra=False
326  for (n,dir) in directories.items():
327  chainDict=copy.deepcopy(self.defaultChain)
328  print("inspecting",dir)
329  nextHasDSInput=None
330  for (x,s) in mReader.workFlowSteps.items():
331  #x has the format (num, prefix)
332  #s has the format (num, name, commands, stepList)
333  if x[0]==n:
334  #print "found",n,s[3]
335  #chainDict['RequestString']='RV'+chainDict['CMSSWVersion']+s[1].split('+')[0]
336  index=0
337  splitForThisWf=None
338  thisLabel=self.speciallabel
339  #if 'HARVESTGEN' in s[3]:
340  if len( [step for step in s[3] if "HARVESTGEN" in step] )>0:
341  chainDict['TimePerEvent']=0.01
342  thisLabel=thisLabel+"_gen"
343  # for double miniAOD test
344  if len( [step for step in s[3] if "DBLMINIAODMCUP15NODQM" in step] )>0:
345  thisLabel=thisLabel+"_dblMiniAOD"
346  processStrPrefix=''
347  setPrimaryDs=None
348  nanoedmGT=''
349  for step in s[3]:
350 
351  if 'INPUT' in step or (not isinstance(s[2][index],str)):
352  nextHasDSInput=s[2][index]
353 
354  else:
355 
356  if (index==0):
357  #first step and not input -> gen part
358  chainDict['nowmTasklist'].append(copy.deepcopy(self.defaultScratch))
359  try:
360  chainDict['nowmTasklist'][-1]['nowmIO']=json.loads(open('%s/%s.io'%(dir,step)).read())
361  except:
362  print("Failed to find",'%s/%s.io'%(dir,step),".The workflows were probably not run on cfg not created")
363  return -15
364 
365  chainDict['nowmTasklist'][-1]['PrimaryDataset']='RelVal'+s[1].split('+')[0]
366  if not '--relval' in s[2][index]:
367  print('Impossible to create task from scratch without splitting information with --relval')
368  return -12
369  else:
370  arg=s[2][index].split()
371  ns=list(map(int,arg[len(arg) - arg[-1::-1].index('--relval')].split(',')))
372  chainDict['nowmTasklist'][-1]['RequestNumEvents'] = ns[0]
373  chainDict['nowmTasklist'][-1]['EventsPerJob'] = ns[1]
374  chainDict['nowmTasklist'][-1]['EventsPerLumi'] = ns[1]
375  #overwrite EventsPerLumi if numberEventsInLuminosityBlock is set in cmsDriver
376  if 'numberEventsInLuminosityBlock' in s[2][index]:
377  nEventsInLuminosityBlock = re.findall('process.source.numberEventsInLuminosityBlock=cms.untracked.uint32\(([ 0-9 ]*)\)', s[2][index],re.DOTALL)
378  if nEventsInLuminosityBlock[-1].isdigit() and int(nEventsInLuminosityBlock[-1]) < ns[1]:
379  chainDict['nowmTasklist'][-1]['EventsPerLumi'] = int(nEventsInLuminosityBlock[-1])
380  if(self.numberEventsInLuminosityBlock > 0 and self.numberEventsInLuminosityBlock <= ns[1]):
381  chainDict['nowmTasklist'][-1]['EventsPerLumi'] = self.numberEventsInLuminosityBlock
382  if 'FASTSIM' in s[2][index] or '--fast' in s[2][index]:
383  thisLabel+='_FastSim'
384  if 'lhe' in s[2][index] in s[2][index]:
385  chainDict['nowmTasklist'][-1]['LheInputFiles'] =True
386 
387  elif nextHasDSInput:
388  chainDict['nowmTasklist'].append(copy.deepcopy(self.defaultInput))
389  try:
390  chainDict['nowmTasklist'][-1]['nowmIO']=json.loads(open('%s/%s.io'%(dir,step)).read())
391  except:
392  print("Failed to find",'%s/%s.io'%(dir,step),".The workflows were probably not run on cfg not created")
393  return -15
394  chainDict['nowmTasklist'][-1]['InputDataset']=nextHasDSInput.dataSet
395  if ('DQMHLTonRAWAOD' in step) :
396  chainDict['nowmTasklist'][-1]['IncludeParents']=True
397  splitForThisWf=nextHasDSInput.split
398  chainDict['nowmTasklist'][-1]['LumisPerJob']=splitForThisWf
399  if step in wmsplit:
400  chainDict['nowmTasklist'][-1]['LumisPerJob']=wmsplit[step]
401  # get the run numbers or #events
402  if len(nextHasDSInput.run):
403  chainDict['nowmTasklist'][-1]['RunWhitelist']=nextHasDSInput.run
404  if len(nextHasDSInput.ls):
405  chainDict['nowmTasklist'][-1]['LumiList']=nextHasDSInput.ls
406  #print "what is s",s[2][index]
407  if '--data' in s[2][index] and nextHasDSInput.label:
408  thisLabel+='_RelVal_%s'%nextHasDSInput.label
409  if 'filter' in chainDict['nowmTasklist'][-1]['nowmIO']:
410  print("This has an input DS and a filter sequence: very likely to be the PyQuen sample")
411  processStrPrefix='PU_'
412  setPrimaryDs = 'RelVal'+s[1].split('+')[0]
413  if setPrimaryDs:
414  chainDict['nowmTasklist'][-1]['PrimaryDataset']=setPrimaryDs
415  nextHasDSInput=None
416  else:
417  #not first step and no inputDS
418  chainDict['nowmTasklist'].append(copy.deepcopy(self.defaultTask))
419  try:
420  chainDict['nowmTasklist'][-1]['nowmIO']=json.loads(open('%s/%s.io'%(dir,step)).read())
421  except:
422  print("Failed to find",'%s/%s.io'%(dir,step),".The workflows were probably not run on cfg not created")
423  return -15
424  if splitForThisWf:
425  chainDict['nowmTasklist'][-1]['LumisPerJob']=splitForThisWf
426  if step in wmsplit:
427  chainDict['nowmTasklist'][-1]['LumisPerJob']=wmsplit[step]
428 
429  # change LumisPerJob for Hadronizer steps.
430  if 'Hadronizer' in step:
431  chainDict['nowmTasklist'][-1]['LumisPerJob']=wmsplit['Hadronizer']
432 
433  #print step
434  chainDict['nowmTasklist'][-1]['TaskName']=step
435  if setPrimaryDs:
436  chainDict['nowmTasklist'][-1]['PrimaryDataset']=setPrimaryDs
437  chainDict['nowmTasklist'][-1]['ConfigCacheID']='%s/%s.py'%(dir,step)
438  chainDict['nowmTasklist'][-1]['GlobalTag']=chainDict['nowmTasklist'][-1]['nowmIO']['GT'] # copy to the proper parameter name
439  chainDict['GlobalTag']=chainDict['nowmTasklist'][-1]['nowmIO']['GT'] #set in general to the last one of the chain
440  if 'NANOEDM' in step :
441  nanoedmGT = chainDict['nowmTasklist'][-1]['nowmIO']['GT']
442  if 'NANOMERGE' in step :
443  chainDict['GlobalTag'] = nanoedmGT
444  if 'pileup' in chainDict['nowmTasklist'][-1]['nowmIO']:
445  chainDict['nowmTasklist'][-1]['MCPileup']=chainDict['nowmTasklist'][-1]['nowmIO']['pileup']
446  if '--pileup ' in s[2][index]: # catch --pileup (scenarion) and not --pileup_ (dataset to be mixed) => works also making PRE-MIXed dataset
447  processStrPrefix='PU_' # take care of pu overlay done with GEN-SIM mixing
448  if ( s[2][index].split()[ s[2][index].split().index('--pileup')+1 ] ).find('25ns') > 0 :
449  processStrPrefix='PU25ns_'
450  elif ( s[2][index].split()[ s[2][index].split().index('--pileup')+1 ] ).find('50ns') > 0 :
451  processStrPrefix='PU50ns_'
452  if 'premix_stage2' in s[2][index] and '--pileup_input' in s[2][index]: # take care of pu overlay done with DIGI mixing of premixed events
453  if s[2][index].split()[ s[2][index].split().index('--pileup_input')+1 ].find('25ns') > 0 :
454  processStrPrefix='PUpmx25ns_'
455  elif s[2][index].split()[ s[2][index].split().index('--pileup_input')+1 ].find('50ns') > 0 :
456  processStrPrefix='PUpmx50ns_'
457 
458  if acqEra:
459  #chainDict['AcquisitionEra'][step]=(chainDict['CMSSWVersion']+'-PU_'+chainDict['nowmTasklist'][-1]['GlobalTag']).replace('::All','')+thisLabel
460  chainDict['AcquisitionEra'][step]=chainDict['CMSSWVersion']
461  chainDict['ProcessingString'][step]=processStrPrefix+chainDict['nowmTasklist'][-1]['GlobalTag'].replace('::All','').replace('-','_')+thisLabel
462  if 'NANOMERGE' in step :
463  chainDict['ProcessingString'][step]=processStrPrefix+nanoedmGT.replace('::All','').replace('-','_')+thisLabel
464  else:
465  #chainDict['nowmTasklist'][-1]['AcquisitionEra']=(chainDict['CMSSWVersion']+'-PU_'+chainDict['nowmTasklist'][-1]['GlobalTag']).replace('::All','')+thisLabel
466  chainDict['nowmTasklist'][-1]['AcquisitionEra']=chainDict['CMSSWVersion']
467  chainDict['nowmTasklist'][-1]['ProcessingString']=processStrPrefix+chainDict['nowmTasklist'][-1]['GlobalTag'].replace('::All','').replace('-','_')+thisLabel
468  if 'NANOMERGE' in step :
469  chainDict['nowmTasklist'][-1]['ProcessingString']=processStrPrefix+nanoedmGT.replace('::All','').replace('-','_')+thisLabel
470 
471  if (self.batchName):
472  chainDict['nowmTasklist'][-1]['Campaign'] = chainDict['nowmTasklist'][-1]['AcquisitionEra']+self.batchName
473 
474  # specify different ProcessingString for double miniAOD dataset
475  if ('DBLMINIAODMCUP15NODQM' in step):
476  chainDict['nowmTasklist'][-1]['ProcessingString']=chainDict['nowmTasklist'][-1]['ProcessingString']+'_miniAOD'
477 
478  if( chainDict['nowmTasklist'][-1]['Multicore'] ):
479  # the scaling factor of 1.2GB / thread is empirical and measured on a SECOND round of tests with PU samples
480  # the number of threads is NO LONGER assumed to be the same for all tasks
481  # https://hypernews.cern.ch/HyperNews/CMS/get/edmFramework/3509/1/1/1.html
482  # now change to 1.5GB / additional thread according to discussion:
483  # https://hypernews.cern.ch/HyperNews/CMS/get/relval/4817/1/1.html
484 # chainDict['nowmTasklist'][-1]['Memory'] = 3000 + int( chainDict['nowmTasklist'][-1]['Multicore'] -1 )*1500
485  chainDict['nowmTasklist'][-1]['Memory'] = self.memoryOffset + int( chainDict['nowmTasklist'][-1]['Multicore'] -1 ) * self.memPerCore
486 
487  index+=1
488  #end of loop through steps
489  chainDict['RequestString']='RV'+chainDict['CMSSWVersion']+s[1].split('+')[0]
490  if processStrPrefix or thisLabel:
491  chainDict['RequestString']+='_'+processStrPrefix+thisLabel
492  #check candidate WF name
493  self.candidateWFName = self.user+'_'+chainDict['RequestString']
494  if (len(self.candidateWFName)>MAXWORKFLOWLENGTH):
495  self.longWFName.append(self.candidateWFName)
496 
497 
498  chainDict['PrepID'] = chainDict['CMSSWVersion']+'__'+self.batchTime+'-'+s[1].split('+')[0]
499  if(self.batchName):
500  chainDict['PrepID'] = chainDict['CMSSWVersion']+self.batchName+'-'+s[1].split('+')[0]
501  if( 'HIN' in self.batchName ):
502  chainDict['SubRequestType'] = "HIRelVal"
503 
504  #wrap up for this one
505  import pprint
506  #print 'wrapping up'
507  #pprint.pprint(chainDict)
508  #loop on the task list
509  for i_second in reversed(range(len(chainDict['nowmTasklist']))):
510  t_second=chainDict['nowmTasklist'][i_second]
511  #print "t_second taskname", t_second['TaskName']
512  if 'primary' in t_second['nowmIO']:
513  #print t_second['nowmIO']['primary']
514  primary=t_second['nowmIO']['primary'][0].replace('file:','')
515  for i_input in reversed(range(0,i_second)):
516  t_input=chainDict['nowmTasklist'][i_input]
517  for (om,o) in t_input['nowmIO'].items():
518  if primary in o:
519  #print "found",primary,"procuced by",om,"of",t_input['TaskName']
520  #ad-hoc fix due to restriction in TaskName of 50 characters
521  if (len(t_input['TaskName'])>50):
522  if (t_input['TaskName'].find('GenSim') != -1):
523  t_input['TaskName'] = 'GenSimFull'
524  if (t_input['TaskName'].find('Hadronizer') != -1):
525  t_input['TaskName'] = 'HadronizerFull'
526  t_second['InputTask'] = t_input['TaskName']
527  t_second['InputFromOutputModule'] = om
528  #print 't_second',pprint.pformat(t_second)
529  if t_second['TaskName'].startswith('HARVEST'):
530  chainDict.update(copy.deepcopy(self.defaultHarvest))
531  if "_RD" in t_second['TaskName']:
532  chainDict['DQMHarvestUnit'] = "multiRun"
533  chainDict['DQMConfigCacheID']=t_second['ConfigCacheID']
534 
537  break
538 
539  # agreed changes for wm injection:
540  # - Campaign: *optional* string during creation. It will default to AcqEra value if possible.
541  # Otherwise it will be empty.
542  # - AcquisitionEra: *mandatory* string at request level during creation. *optional* string
543  # at task level during creation. "optional" during assignment.
544  # - ProcessingString: *mandatory* string at request level during creation. *optional* string
545  # at task level during creation. "optional" during assignment.
546  # - ProcessingVersion: *optional* during creation (default 1). *optional* during assignment.
547  #
548  # Which requires following changes here:
549  # - reset Global AcuisitionEra, ProcessingString to be the one in the first task
550  # - and also Campaign to be always the same as the AcquisitionEra
551 
552  if acqEra:
553  chainDict['AcquisitionEra'] = chainDict['AcquisitionEra'].values()[0]
554  chainDict['ProcessingString'] = chainDict['ProcessingString'].values()[0]
555  else:
556  chainDict['AcquisitionEra'] = chainDict['nowmTasklist'][0]['AcquisitionEra']
557  chainDict['ProcessingString'] = chainDict['nowmTasklist'][0]['ProcessingString']
558 
559 
561  chainDict['Campaign'] = chainDict['AcquisitionEra']+self.batchName
562 
563 
564  itask=0
565  if self.keep:
566  for i in self.keep:
567  if isinstance(i, int) and i < len(chainDict['nowmTasklist']):
568  chainDict['nowmTasklist'][i]['KeepOutput']=True
569  for (i,t) in enumerate(chainDict['nowmTasklist']):
570  if t['TaskName'].startswith('HARVEST'):
571  continue
572  if not self.keep:
573  t['KeepOutput']=True
574  elif t['TaskName'] in self.keep:
575  t['KeepOutput']=True
576  if t['TaskName'].startswith('HYBRIDRepackHI2015VR'):
577  t['KeepOutput']=False
578  t.pop('nowmIO')
579  itask+=1
580  chainDict['Task%d'%(itask)]=t
581 
582 
583 
584 
585 
586 
587  chainDict['TaskChain']=itask#len(chainDict['nowmTasklist'])
588 
589  chainDict.pop('nowmTasklist')
590  self.chainDicts[n]=chainDict
591 
592 
593  return 0
594 
595  def uploadConf(self,filePath,label,where):
596  labelInCouch=self.label+'_'+label
597  cacheName=filePath.split('/')[-1]
598  if self.testMode:
599  self.count+=1
600  print('\tFake upload of',filePath,'to couch with label',labelInCouch)
601  return self.count
602  else:
603  try:
604  from modules.wma import upload_to_couch,DATABASE_NAME
605  except:
606  print('\n\tUnable to find wmcontrol modules. Please include it in your python path\n')
607  print('\n\t QUIT\n')
608  sys.exit(-16)
609 
610  if cacheName in self.couchCache:
611  print("Not re-uploading",filePath,"to",where,"for",label)
612  cacheId=self.couchCache[cacheName]
613  else:
614  print("Loading",filePath,"to",where,"for",label)
615 
616  pool = multiprocessing.Pool(1)
617  cacheIds = pool.map( upload_to_couch_oneArg, [(filePath,labelInCouch,self.user,self.group,where)] )
618  cacheId = cacheIds[0]
619  self.couchCache[cacheName]=cacheId
620  return cacheId
621 
622  def upload(self):
623  for (n,d) in self.chainDicts.items():
624  for it in d:
625  if it.startswith("Task") and it!='TaskChain':
626  #upload
627  couchID=self.uploadConf(d[it]['ConfigCacheID'],
628  str(n)+d[it]['TaskName'],
629  d['ConfigCacheUrl']
630  )
631  print(d[it]['ConfigCacheID']," uploaded to couchDB for",str(n),"with ID",couchID)
632  d[it]['ConfigCacheID']=couchID
633  if it =='DQMConfigCacheID':
634  couchID=self.uploadConf(d['DQMConfigCacheID'],
635  str(n)+'harvesting',
636  d['ConfigCacheUrl']
637  )
638  print(d['DQMConfigCacheID'],"uploaded to couchDB for",str(n),"with ID",couchID)
639  d['DQMConfigCacheID']=couchID
640 
641 
642  def submit(self):
643  try:
644  from modules.wma import makeRequest,approveRequest
645  from wmcontrol import random_sleep
646  print('\n\tFound wmcontrol\n')
647  except:
648  print('\n\tUnable to find wmcontrol modules. Please include it in your python path\n')
649  if not self.testMode:
650  print('\n\t QUIT\n')
651  sys.exit(-17)
652 
653  import pprint
654  for (n,d) in self.chainDicts.items():
655  if self.testMode:
656  print("Only viewing request",n)
657  print(pprint.pprint(d))
658  else:
659  #submit to wmagent each dict
660  print("For eyes before submitting",n)
661  print(pprint.pprint(d))
662  print("Submitting",n,"...........")
663  workFlow=makeRequest(self.wmagent,d,encodeDict=True)
664  print("...........",n,"submitted")
665  random_sleep()
666  if self.testMode and len(self.longWFName)>0:
667  print("\n*** WARNING: "+str(len(self.longWFName))+" workflows have too long names for submission (>"+str(MAXWORKFLOWLENGTH)+ "characters) ***")
668  print('\n'.join(self.longWFName))
MatrixInjector.MatrixInjector.upload
def upload(self)
Definition: MatrixInjector.py:622
MatrixInjector.MatrixInjector.longWFName
longWFName
Definition: MatrixInjector.py:88
FastTimerService_cff.range
range
Definition: FastTimerService_cff.py:34
resolutioncreator_cfi.object
object
Definition: resolutioncreator_cfi.py:4
MatrixInjector.MatrixInjector
Definition: MatrixInjector.py:41
MatrixInjector.MatrixInjector.defaultInput
defaultInput
Definition: MatrixInjector.py:143
MatrixInjector.MatrixInjector.keep
keep
Definition: MatrixInjector.py:56
MatrixInjector.MatrixInjector.memoryOffset
memoryOffset
Definition: MatrixInjector.py:57
MatrixInjector.MatrixInjector.couch
couch
Definition: MatrixInjector.py:79
join
static std::string join(char **cmd)
Definition: RemoteFile.cc:17
MatrixInjector.MatrixInjector.DbsUrl
DbsUrl
Definition: MatrixInjector.py:71
MatrixInjector.MatrixInjector.couchCache
couchCache
Definition: MatrixInjector.py:81
cms::dd::split
std::vector< std::string_view > split(std::string_view, const char *)
spr::find
void find(edm::Handle< EcalRecHitCollection > &hits, DetId thisDet, std::vector< EcalRecHitCollection::const_iterator > &hit, bool debug=false)
Definition: FindCaloHit.cc:19
MatrixInjector.upload_to_couch_oneArg
def upload_to_couch_oneArg(arguments)
Definition: MatrixInjector.py:29
mps_monitormerge.items
list items
Definition: mps_monitormerge.py:29
MatrixInjector.MatrixInjector.wmagent
wmagent
Definition: MatrixInjector.py:47
MatrixInjector.MatrixInjector.batchTime
batchTime
Definition: MatrixInjector.py:61
MatrixInjector.MatrixInjector.candidateWFName
candidateWFName
Definition: MatrixInjector.py:493
MatrixInjector.MatrixInjector.speciallabel
speciallabel
Definition: MatrixInjector.py:85
contentValuesCheck.values
values
Definition: contentValuesCheck.py:38
MatrixInjector.MatrixInjector.uploadConf
def uploadConf(self, filePath, label, where)
the info are not in the task specific dict but in the general dict t_input.update(copy....
Definition: MatrixInjector.py:595
str
#define str(s)
Definition: TestProcessor.cc:48
MatrixInjector.MatrixInjector.defaultHarvest
defaultHarvest
Definition: MatrixInjector.py:122
MatrixInjector.MatrixInjector.prepare
def prepare(self, mReader, directories, mode='init')
Definition: MatrixInjector.py:323
MatrixInjector.MatrixInjector.group
group
Definition: MatrixInjector.py:83
MatrixInjector.MatrixInjector.chainDicts
chainDicts
Definition: MatrixInjector.py:167
MatrixInjector.MatrixInjector.defaultTask
defaultTask
Definition: MatrixInjector.py:154
MatrixInjector.MatrixInjector.testMode
testMode
Definition: MatrixInjector.py:54
mps_setup.append
append
Definition: mps_setup.py:85
MatrixInjector.MatrixInjector.batchName
batchName
Definition: MatrixInjector.py:60
MatrixInjector.MatrixInjector.count
count
Definition: MatrixInjector.py:44
MatrixInjector.MatrixInjector.defaultScratch
defaultScratch
Definition: MatrixInjector.py:129
createfilelist.int
int
Definition: createfilelist.py:10
MatrixInjector.MatrixInjector.memPerCore
memPerCore
Definition: MatrixInjector.py:58
MatrixInjector.MatrixInjector.__init__
def __init__(self, opt, mode='init', options='')
Definition: MatrixInjector.py:43
MatrixInjector.performInjectionOptionTest
def performInjectionOptionTest(opt)
Definition: MatrixInjector.py:12
MatrixInjector.MatrixInjector.submit
def submit(self)
Definition: MatrixInjector.py:642
edm::print
S & print(S &os, JobReport::InputFile const &f)
Definition: JobReport.cc:66
MatrixInjector.MatrixInjector.get_wmsplit
def get_wmsplit()
Definition: MatrixInjector.py:171
readEcalDQMStatus.read
read
Definition: readEcalDQMStatus.py:38
MatrixInjector.MatrixInjector.label
label
Definition: MatrixInjector.py:84
MatrixInjector.MatrixInjector.numberEventsInLuminosityBlock
numberEventsInLuminosityBlock
Definition: MatrixInjector.py:59
AlignmentPI::index
index
Definition: AlignmentPayloadInspectorHelper.h:46
list
How EventSelector::AcceptEvent() decides whether to accept an event for output otherwise it is excluding the probing of A single or multiple positive and the trigger will pass if any such matching triggers are PASS or EXCEPTION[A criterion thatmatches no triggers at all is detected and causes a throw.] A single negative with an expectation of appropriate bit checking in the decision and the trigger will pass if any such matching triggers are FAIL or EXCEPTION A wildcarded negative criterion that matches more than one trigger in the trigger list("!*", "!HLTx*" if it matches 2 triggers or more) will accept the event if all the matching triggers are FAIL. It will reject the event if any of the triggers are PASS or EXCEPTION(this matches the behavior of "!*" before the partial wildcard feature was incorporated). Triggers which are in the READY state are completely ignored.(READY should never be returned since the trigger paths have been run
genParticles_cff.map
map
Definition: genParticles_cff.py:11
MatrixInjector.MatrixInjector.version
version
Definition: MatrixInjector.py:55
MatrixInjector.MatrixInjector.user
user
Definition: MatrixInjector.py:82
MatrixInjector.MatrixInjector.defaultChain
defaultChain
Definition: MatrixInjector.py:98
MatrixInjector.MatrixInjector.dqmgui
dqmgui
Definition: MatrixInjector.py:46
python.rootplot.root2matplotlib.replace
def replace(string, replacements)
Definition: root2matplotlib.py:444
Generated for CMSSW Reference Manual by doxygen 1.8.16