File indexing completed on 2021-02-14 12:47:44
0001 from __future__ import print_function
0002 from __future__ import absolute_import
0003 import os,sys
0004 import glob
0005 import logging
0006 import argparse
0007 import subprocess
0008 import time, datetime
0009 import urllib2
0010 import json
0011
0012 from . import tools
0013 from .CLIHelper import CLIHelper
0014 from .CrabHelper import CrabHelper
0015 import FWCore.ParameterSet.Config as cms
0016 log = logging.getLogger(__name__)
0017
0018 class DTWorkflow(CLIHelper, CrabHelper):
0019 """ This is the base class for all DTWorkflows and contains some
0020 common tasks """
0021 def __init__(self, options):
0022 self.options = options
0023 super( DTWorkflow, self ).__init__()
0024 self.digilabel = "muonDTDigis"
0025
0026
0027 self.required_options_dict = {}
0028 self.required_options_prepare_dict = {}
0029 self.fill_required_options_dict()
0030 self.fill_required_options_prepare_dict()
0031
0032 self.pset_name = ""
0033 self.outpath_command_tag = ""
0034 self.output_files = []
0035 self.input_files = []
0036
0037 self.run_all_command = False
0038 self.files_reveived = False
0039 self._user = ""
0040
0041 os.chdir(self.options.working_dir)
0042
0043 def check_missing_options(self, requirements_dict):
0044 missing_options = []
0045
0046 if self.options.command in requirements_dict:
0047 for option in requirements_dict[self.options.command]:
0048 if not (hasattr(self.options, option)
0049 and ( (getattr(self.options,option))
0050 or isinstance(getattr(self.options,option), bool) )):
0051 missing_options.append(option)
0052 if len(missing_options) > 0:
0053 err = "The following CLI options are missing"
0054 err += " for command %s: " % self.options.command
0055 err += " ".join(missing_options)
0056 raise ValueError(err)
0057
0058 def run(self):
0059 """ Generalized function to run workflow command"""
0060 msg = "Preparing %s workflow" % self.options.workflow
0061 if hasattr(self.options, "command"):
0062 msg += " for command %s" % self.options.command
0063 log.info(msg)
0064 if self.options.config_path:
0065 self.load_options( self.options.config_path )
0066
0067 self.check_missing_options(self.required_options_prepare_dict)
0068 self.prepare_workflow()
0069
0070 if not os.path.exists( self.local_path ):
0071 os.makedirs(self.local_path)
0072
0073 self.dump_options()
0074
0075 self.check_missing_options(self.required_options_dict)
0076 try:
0077 run_function = getattr(self, self.options.command)
0078 except AttributeError:
0079 errmsg = "Class `{}` does not implement `{}` for workflow %s" % self.options.workflow
0080 if hasattr(self.options, "workflow_mode"):
0081 errmsg += "and workflow mode %s" % self.options.workflow_mode
0082 raise NotImplementedError( errmsg.format(self.__class__.__name__,
0083 self.options.command))
0084 log.debug("Running command %s" % self.options.command)
0085
0086 run_function()
0087
0088 def prepare_workflow(self):
0089 """ Abstract implementation of prepare workflow function"""
0090 errmsg = "Class `{}` does not implement `{}`"
0091 raise NotImplementedError( errmsg.format(self.__class__.__name__,
0092 "prepare_workflow"))
0093
0094 def all(self):
0095 """ generalized function to perform several workflow mode commands in chain.
0096 All commands mus be specified in self.all_commands list in workflow mode specific
0097 prepare function in child workflow objects.
0098 """
0099 self.run_all_command = True
0100 for command in self.all_commands:
0101 self.options.command = command
0102 self.run()
0103
0104 def submit(self):
0105 self.submit_crab_task()
0106
0107 def check(self):
0108 """ Function to check status of submitted tasks """
0109 self.check_crabtask()
0110
0111 def write(self):
0112 self.runCMSSWtask()
0113
0114 def dump(self):
0115 self.runCMSSWtask()
0116
0117 def correction(self):
0118 self.runCMSSWtask()
0119
0120 def add_preselection(self):
0121 """ Add preselection to the process object stored in workflow_object"""
0122 if not hasattr(self, "process"):
0123 raise NameError("Process is not initalized in workflow object")
0124 pathsequence = self.options.preselection.split(':')[0]
0125 seqname = self.options.preselection.split(':')[1]
0126 self.process.load(pathsequence)
0127 tools.prependPaths(self.process, seqname)
0128
0129 def add_raw_option(self):
0130 getattr(self.process, self.digilabel).inputLabel = 'rawDataCollector'
0131 tools.prependPaths(self.process,self.digilabel)
0132
0133 def add_local_t0_db(self, local=False):
0134 """ Add a local t0 database as input. Use the option local is used
0135 if the pset is processed locally and not with crab.
0136 """
0137 if local:
0138 connect = os.path.abspath(self.options.inputT0DB)
0139 else:
0140 connect = os.path.basename(self.options.inputT0DB)
0141 self.addPoolDBESSource( process = self.process,
0142 moduleName = 't0DB',
0143 record = 'DTT0Rcd',
0144 tag = 't0',
0145 connect = 'sqlite_file:%s' % connect)
0146 self.input_files.append(os.path.abspath(self.options.inputT0DB))
0147
0148 def add_local_vdrift_db(self, local=False):
0149 """ Add a local vdrift database as input. Use the option local is used
0150 if the pset is processed locally and not with crab.
0151 """
0152 if local:
0153 connect = os.path.abspath(self.options.inputVDriftDB)
0154 else:
0155 connect = os.path.basename(self.options.inputVDriftDB)
0156 self.addPoolDBESSource( process = self.process,
0157 moduleName = 'vDriftDB',
0158 record = 'DTMtimeRcd',
0159 tag = 'vDrift',
0160 connect = 'sqlite_file:%s' % connect)
0161 self.input_files.append( os.path.abspath(self.options.inputVDriftDB) )
0162
0163 def add_local_calib_db(self, local=False):
0164 """ Add a local calib database as input. Use the option local is used
0165 if the pset is processed locally and not with crab.
0166 """
0167 label = ''
0168 if self.options.datasettype == "Cosmics":
0169 label = 'cosmics'
0170 if local:
0171 connect = os.path.abspath(self.options.inputCalibDB)
0172 else:
0173 connect = os.path.basename(self.options.inputCalibDB)
0174 self.addPoolDBESSource( process = self.process,
0175 moduleName = 'calibDB',
0176 record = 'DTTtrigRcd',
0177 tag = 'ttrig',
0178 connect = str("sqlite_file:%s" % connect),
0179 label = label
0180 )
0181 self.input_files.append( os.path.abspath(self.options.inputCalibDB) )
0182
0183 def add_local_custom_db(self):
0184 for option in ('inputDBRcd', 'connectStrDBTag'):
0185 if hasattr(self.options, option) and not getattr(self.options, option):
0186 raise ValueError("Option %s needed for custom input db" % option)
0187 self.addPoolDBESSource( process = self.process,
0188 record = self.options.inputDBRcd,
0189 tag = self.options.inputDBTag,
0190 connect = self.options.connectStrDBTag,
0191 moduleName = 'customDB%s' % self.options.inputDBRcd
0192 )
0193
0194 def prepare_common_submit(self):
0195 """ Common operations used in most prepare_[workflow_mode]_submit functions"""
0196 if not self.options.run:
0197 raise ValueError("Option run is required for submission!")
0198 if hasattr(self.options, "inputT0DB") and self.options.inputT0DB:
0199 self.add_local_t0_db()
0200
0201 if hasattr(self.options, "inputVDriftDB") and self.options.inputVDriftDB:
0202 self.add_local_vdrift_db()
0203
0204 if hasattr(self.options, "inputDBTag") and self.options.inputDBTag:
0205 self.add_local_custom_db()
0206
0207 if self.options.run_on_RAW:
0208 self.add_raw_option()
0209 if self.options.preselection:
0210 self.add_preselection()
0211
0212 def prepare_common_write(self, do_hadd=True):
0213 """ Common operations used in most prepare_[workflow_mode]_erite functions"""
0214 self.load_options_command("submit")
0215 output_path = os.path.join( self.local_path, "unmerged_results" )
0216 merged_file = os.path.join(self.result_path, self.output_file)
0217 crabtask = self.crabFunctions.CrabTask(crab_config = self.crab_config_filepath,
0218 initUpdate = False)
0219 if not (self.options.skip_stageout or self.files_reveived or self.options.no_exec):
0220 self.get_output_files(crabtask, output_path)
0221 log.info("Received files from storage element")
0222 log.info("Using hadd to merge output files")
0223 if not self.options.no_exec and do_hadd:
0224 returncode = tools.haddLocal(output_path, merged_file)
0225 if returncode != 0:
0226 raise RuntimeError("Failed to merge files with hadd")
0227 return crabtask.crabConfig.Data.outputDatasetTag
0228
0229 def prepare_common_dump(self, db_path):
0230 self.process = tools.loadCmsProcess(self.pset_template)
0231 self.process.calibDB.connect = 'sqlite_file:%s' % db_path
0232 try:
0233 path = self.result_path
0234 except:
0235 path = os.getcwd()
0236 print("path", path)
0237 out_path = os.path.abspath(os.path.join(path,
0238 os.path.splitext(db_path)[0] + ".txt"))
0239
0240 self.process.dumpToFile.outputFileName = out_path
0241
0242 @staticmethod
0243 def addPoolDBESSource( process,
0244 moduleName,
0245 record,
0246 tag,
0247 connect='sqlite_file:',
0248 label='',):
0249
0250 from CondCore.CondDB.CondDB_cfi import CondDB
0251
0252 calibDB = cms.ESSource("PoolDBESSource",
0253 CondDB,
0254 timetype = cms.string('runnumber'),
0255 toGet = cms.VPSet(cms.PSet(
0256 record = cms.string(record),
0257 tag = cms.string(tag),
0258 label = cms.untracked.string(label)
0259 )),
0260 )
0261 calibDB.connect = cms.string( str(connect) )
0262
0263 if 'oracle:' in connect:
0264 calibDB.DBParameters.authenticationPath = '/afs/cern.ch/cms/DB/conddb'
0265 setattr(process,moduleName,calibDB)
0266 setattr(process,"es_prefer_" + moduleName,cms.ESPrefer('PoolDBESSource',
0267 moduleName)
0268 )
0269
0270 def get_output_files(self, crabtask, output_path):
0271 self.crab.callCrabCommand( ["getoutput",
0272 "--outputpath",
0273 output_path,
0274 crabtask.crabFolder ] )
0275
0276 def runCMSSWtask(self, pset_path=""):
0277 """ Run a cmsRun job locally. The member variable self.pset_path is used
0278 if pset_path argument is not given"""
0279 if self.options.no_exec:
0280 return 0
0281 process = subprocess.Popen( "cmsRun %s" % self.pset_path,
0282 stdout=subprocess.PIPE,
0283 stderr=subprocess.STDOUT,
0284 shell = True)
0285 stdout = process.communicate()[0]
0286 log.info(stdout)
0287 if process.returncode != 0:
0288 raise RuntimeError("Failed to use cmsRun for pset %s" % self.pset_name)
0289 return process.returncode
0290
0291 @property
0292 def remote_out_path(self):
0293 """ Output path on remote excluding user base path
0294 Returns a dict if crab is used due to crab path setting policy"""
0295 if self.options.command =="submit":
0296 return {
0297 "outLFNDirBase" : os.path.join( "/store",
0298 "user",
0299 self.user,
0300 'DTCalibration/',
0301 self.outpath_command_tag,
0302 self.outpath_workflow_mode_tag),
0303 "outputDatasetTag" : self.tag
0304 }
0305 else:
0306 return os.path.join( 'DTCalibration/',
0307 datasetstr,
0308 'Run' + str(self.options.run),
0309 self.outpath_command_tag,
0310 self.outpath_workflow_mode_tag,
0311 'v' + str(self.options.trial),
0312 )
0313 @property
0314 def outpath_workflow_mode_tag(self):
0315 if not self.options.workflow_mode in self.outpath_workflow_mode_dict:
0316 raise NotImplementedError("%s missing in outpath_workflow_mode_dict" % self.options.workflow_mode)
0317 return self.outpath_workflow_mode_dict[self.options.workflow_mode]
0318
0319 @property
0320 def tag(self):
0321 return 'Run' + str(self.options.run) + '_v' + str(self.options.trial)
0322
0323 @property
0324 def user(self):
0325 if self._user:
0326 return self._user
0327 if hasattr(self.options, "user") and self.options.user:
0328 self._user = self.options.user
0329 else:
0330 self._user = self.crab.checkusername()
0331 return self._user
0332
0333 @property
0334 def local_path(self):
0335 """ Output path on local machine """
0336 if self.options.run and self.options.label:
0337 prefix = "Run%d-%s_v%d" % ( self.options.run,
0338 self.options.label,
0339 self.options.trial)
0340 else:
0341 prefix = ""
0342 if self.outpath_workflow_mode_tag:
0343 path = os.path.join( self.options.working_dir,
0344 prefix,
0345 self.outpath_workflow_mode_tag)
0346 else:
0347 path = os.path.join( self.options.working_dir,
0348 prefix,
0349 self.outpath_command_tag )
0350 return path
0351
0352 @property
0353 def result_path(self):
0354 result_path = os.path.abspath(os.path.join(self.local_path,"results"))
0355 if not os.path.exists(result_path):
0356 os.makedirs(result_path)
0357 return result_path
0358
0359 @property
0360 def pset_template_base_bath(self):
0361 """ Base path to folder containing pset files for cmsRun"""
0362 return os.path.expandvars(os.path.join("$CMSSW_BASE",
0363 "src",
0364 "CalibMuon",
0365 "test",
0366 )
0367 )
0368
0369 @property
0370 def pset_path(self):
0371 """ full path to the pset file """
0372 basepath = os.path.join( self.local_path, "psets")
0373 if not os.path.exists( basepath ):
0374 os.makedirs( basepath )
0375 return os.path.join( basepath, self.pset_name )
0376
0377 def write_pset_file(self):
0378 if not hasattr(self, "process"):
0379 raise NameError("Process is not initalized in workflow object")
0380 if not os.path.exists(self.local_path):
0381 os.makedirs(self.local_path)
0382 with open( self.pset_path,'w') as pfile:
0383 pfile.write(self.process.dumpPython())
0384
0385 def get_config_name(self, command= ""):
0386 """ Create the name for the output json file which will be dumped"""
0387 if not command:
0388 command = self.options.command
0389 return "config_" + command + ".json"
0390
0391 def dump_options(self):
0392 with open(os.path.join(self.local_path, self.get_config_name()),"w") as out_file:
0393 json.dump(vars(self.options), out_file, indent=4)
0394
0395 def load_options(self, config_file_path):
0396 if not os.path.exists(config_file_path):
0397 raise IOError("File %s not found" % config_file_path)
0398 with open(config_file_path, "r") as input_file:
0399 config_json = json.load(input_file)
0400 for key, val in config_json.items():
0401 if not hasattr(self.options, key) or not getattr(self.options, key):
0402 setattr(self.options, key, val)
0403
0404 def load_options_command(self, command ):
0405 """Load options for previous command in workflow """
0406 if not self.options.config_path:
0407 if not self.options.run:
0408 raise RuntimeError("Option run is required if no config path specified")
0409 if not os.path.exists(self.local_path):
0410 raise IOError("Local path %s does not exist" % self.local_path)
0411 self.options.config_path = os.path.join(self.local_path,
0412 self.get_config_name(command))
0413 self.load_options( self.options.config_path )
0414
