ATLAS Offline Software
Loading...
Searching...
No Matches
trigRecoExe.py
Go to the documentation of this file.
1#!/usr/bin/env python
2
3# Copyright (C) 2002-2026 CERN for the benefit of the ATLAS collaboration
4
5# @brief: Trigger executor to call base transforms
6# @details: Based on athenaExecutor with some modifications
7# @author: Mark Stockton
8
9import os
10import fnmatch
11import re
12import subprocess
13
14from PyJobTransforms.trfExe import athenaExecutor
15
16# imports for preExecute
17from PyJobTransforms.trfUtils import asetupReport, cvmfsDBReleaseCheck, unpackDBRelease, setupDBRelease, lineByLine
18import PyJobTransforms.trfEnv as trfEnv
19import PyJobTransforms.trfExceptions as trfExceptions
20from PyJobTransforms.trfExitCodes import trfExit as trfExit
21import TrigTransform.dbgAnalysis as dbgStream
22from TrigTransform.trigTranslate import getTranslated as getTranslated
23
24# Setup logging here
25import logging, eformat
26msg = logging.getLogger("PyJobTransforms." + __name__)
27
28# Trig_reco_tf.py executor for BS-BS step (aka running the trigger)
29# used to setup input files/arguments and change output filenames
30class trigRecoExecutor(athenaExecutor):
31
32 # Pattern for output files produces by athenaHLT/EF
33 expectedOutputFileName = '*SingleStream.daq.RAW._*.data'
34
35 # preExecute is based on athenaExecutor but with key changes:
36 # - removed athenaMP detection
37 # - removed environment so does not require the noimf notcmalloc flags
38 # - added swap of argument name for runargs file
39 def preExecute(self, input = set(), output = set()):
40 msg.debug('Preparing for execution of {0} with inputs {1} and outputs {2}'.format(self.name, input, output))
41
42 # setsid needed to fix process-group id of child processes to be the same as mother process (ATR-20513)
43 self._exe = 'setsid ' + self.conf.argdict['trigExe'].value
44
45 # Check we actually have events to process!
46 if (self._inputEventTest and 'skipEvents' in self.conf.argdict and
47 self.conf.argdict['skipEvents'].returnMyValue(name=self._name, substep=self._substep, first=self.conf.firstExecutor) is not None):
48 msg.debug('Will test for events to process')
49 for dataType in input:
50 inputEvents = self.conf.dataDictionary[dataType].nentries
51 msg.debug('Got {} events for {}'.format(inputEvents, dataType))
52 if not isinstance(inputEvents, int):
53 msg.warning('Are input events countable? Got nevents={} so disabling event count check for this input'.format(inputEvents))
54 elif self.conf.argdict['skipEvents'].returnMyValue(name=self._name, substep=self._substep, first=self.conf.firstExecutor) >= inputEvents:
55 raise trfExceptions.TransformExecutionException(trfExit.nameToCode('TRF_NOEVENTS'),
56 'No events to process: {0} (skipEvents) >= {1} (inputEvents of {2}'.format(self.conf.argdict['skipEvents'].returnMyValue(name=self._name, substep=self._substep, first=self.conf.firstExecutor), inputEvents, dataType))
57
58
59 if self._skeletonCA is not None:
60 inputFiles = dict()
61 for dataType in input:
62 inputFiles[dataType] = self.conf.dataDictionary[dataType]
63 outputFiles = dict()
64 for dataType in output:
65 outputFiles[dataType] = self.conf.dataDictionary[dataType]
66
67 # See if we have any 'extra' file arguments
68 for dataType, dataArg in self.conf.dataDictionary.items():
69 if dataArg.io == 'input' and self._name in dataArg.executor:
70 inputFiles[dataArg.subtype] = dataArg
71
72 msg.info('Input Files: {0}; Output Files: {1}'.format(inputFiles, outputFiles))
73
74 # Get the list of top options files that will be passed to athena (=runargs file + all skeletons)
75 self._topOptionsFiles = self._jobOptionsTemplate.getTopOptions(input = inputFiles,
76 output = outputFiles)
77
78
80 if len(input) > 0:
81 self._extraMetadata['inputs'] = list(input)
82 if len(output) > 0:
83 self._extraMetadata['outputs'] = list(output)
84
85
86 asetupString = None
87 if 'asetup' in self.conf.argdict:
88 asetupString = self.conf.argdict['asetup'].returnMyValue(name=self._name, substep=self._substep, first=self.conf.firstExecutor)
89 else:
90 msg.info('Asetup report: {0}'.format(asetupReport()))
91
92 # allow overriding the container OS using a flag
93 OSSetupString = None
94 if 'runInContainer' in self.conf.argdict:
95 OSSetupString = self.conf.argdict['runInContainer'].returnMyValue(name=self._name, substep=self._substep, first=self.conf.firstExecutor)
96 msg.info('The step {} will be performed in a container running {}, as explicitly requested'.format(self._substep, OSSetupString))
97
98
99 dbrelease = dbsetup = None
100 if 'DBRelease' in self.conf.argdict:
101 dbrelease = self.conf.argdict['DBRelease'].returnMyValue(name=self._name, substep=self._substep, first=self.conf.firstExecutor)
102 if dbrelease:
103 # Classic tarball - filename format is DBRelease-X.Y.Z.tar.gz
104 dbdMatch = re.match(r'DBRelease-([\d\.]+)\.tar\.gz', os.path.basename(dbrelease))
105 if dbdMatch:
106 msg.debug('DBRelease setting {0} matches classic tarball file'.format(dbrelease))
107 if not os.access(dbrelease, os.R_OK):
108 msg.warning('Transform was given tarball DBRelease file {0}, but this is not there'.format(dbrelease))
109 msg.warning('I will now try to find DBRelease {0} in cvmfs'.format(dbdMatch.group(1)))
110 dbrelease = dbdMatch.group(1)
111 dbsetup = cvmfsDBReleaseCheck(dbrelease)
112 else:
113 # Check if the DBRelease is setup
114 unpacked, dbsetup = unpackDBRelease(tarball=dbrelease, dbversion=dbdMatch.group(1))
115 if unpacked:
116 # Now run the setup.py script to customise the paths to the current location...
117 setupDBRelease(dbsetup)
118 # For cvmfs we want just the X.Y.Z release string (and also support 'current')
119 else:
120 dbsetup = cvmfsDBReleaseCheck(dbrelease)
121
122 # Look for environment updates and perpare the athena command line
124 # above is needed by _prepAthenaCommandLine, but remove the setStandardEnvironment so doesn't include imf or tcmalloc
125 # self._envUpdate.setStandardEnvironment(self.conf.argdict)
126 self._prepAthenaCommandLine()
127
128 # translate relevant parts from the runargs file
129 if 'athenaHLT' in self._exe or 'athenaEF' in self._exe:
130 self._cmd.remove('runargs.BSRDOtoRAW.py')
131 # get list of translated arguments
132 optionList = getTranslated(self.conf.argdict, name=self._name, substep=self._substep, first=self.conf.firstExecutor, output = outputFiles)
133 self._cmd.extend(optionList)
134 # updates for CA
135 if self._isCAEnabled():
136 msg.info("Running in CA mode")
137 # we don't use the runargs file so add the JO and preExecs to the command line
138 self._cmd.append(self._skeletonCA)
139 if 'preExec' in self.conf.argdict:
140 self._cmd.extend(self.conf.argdict['preExec'].returnMyValue(name=self._name, substep=self._substep, first=self.conf.firstExecutor))
141 msg.info('Command adjusted for CA to %s', self._cmd)
142 else:
143 msg.error("Trig_reco_tf does not support running in legacy mode")
144
145 # Run preRun step debug stream analysis if output histogram are set
146 if "outputHIST_DEBUGSTREAMMONFile" in self.conf.argdict:
147 # Do debug stream preRun step and get asetup string from debug stream input files
148 dbgAsetupString, dbAlias = dbgStream.dbgPreRun(self.conf.dataDictionary['BS_RDO'], self.conf.dataDictionary['HIST_DEBUGSTREAMMON'].value, self.conf.argdict)
149 # Setup asetup from debug stream
150 # if no --asetup r2b:string was given and is not running with tzero/software/patches as TestArea
151 if asetupString is None and dbgAsetupString is not None:
152 asetupString = dbgAsetupString
153 msg.info('Will use asetup string for debug stream analysis %s', dbgAsetupString)
154 # allow overriding the container OS using a flag
155 if 'runInContainer' in self.conf.argdict:
156 OSSetupString = self.conf.argdict['runInContainer'].returnMyValue(name=self._name, substep=self._substep, first=self.conf.firstExecutor)
157 msg.info('The step {} will be performed in a container running {}, as explicitly requested'.format(self._substep, OSSetupString))
158
159 # Set database in command line if it was missing
160 if 'useDB' in self.conf.argdict and 'DBserver' not in self.conf.argdict and dbAlias:
161 msg.warn("Database alias will be set to %s", dbAlias)
162 self._cmd.append("--db-server " + dbAlias)
163 else:
164 msg.info("Flag outputHIST_DEBUGSTREAMMONFile not defined - debug stream analysis will not run.")
165
166 # The following is needed to avoid conflicts in finding BS files prduced by running the HLT step
167 # and those already existing in the working directory
168 if 'BS' in self.conf.dataDictionary or 'DRAW_TRIGCOST' in self.conf.dataDictionary or 'HIST_DEBUGSTREAMMON' in self.conf.dataDictionary:
169 # list of filenames of files matching expectedOutputFileName
170 matchedOutputFileNames = self._findOutputFiles(self.expectedOutputFileName)
171 # check there are no file matches
172 if len(matchedOutputFileNames) > 0:
173 msg.error(f'Directoy already contains files with expected output name format ({self.expectedOutputFileName}), please remove/rename these first: {matchedOutputFileNames}')
174 raise trfExceptions.TransformExecutionException(trfExit.nameToCode('TRF_OUTPUT_FILE_ERROR'),
175 f'Directory already contains files with expected output name format {self.expectedOutputFileName}, please remove/rename these first: {matchedOutputFileNames}')
176
177 # Sanity check:
178 if OSSetupString is not None and asetupString is None:
179 raise trfExceptions.TransformExecutionException(trfExit.nameToCode('TRF_EXEC_SETUP_FAIL'),
180 'Athena version must be specified for the substep which requires running inside a container (either via --asetup or from DB)')
181
182 # Call athenaExecutor parent as the above overrides what athenaExecutor would have done
183 super(athenaExecutor, self).preExecute(input, output)
184
185 # Now we always write a wrapper, because it's very convenient for re-running individual substeps
186 # This will have asetup and/or DB release setups in it
187 # Do this last in this preExecute as the _cmd needs to be finalised
188 msg.info('Now writing wrapper for substep executor {0}'.format(self._name))
189 self._writeAthenaWrapper(asetup=asetupString, dbsetup=dbsetup, ossetup=OSSetupString)
190 msg.info('Athena will be executed in a subshell via {0}'.format(self._cmd))
191
192 # Loop over current directory and find the output file matching input pattern
193 def _findOutputFiles(self, pattern):
194 # list to store the filenames of files matching pattern
195 matchedOutputFileNames = []
196 # list of input files that could be in the same folder and need ignoring
197 ignoreInputFileNames = []
198 for dataType, dataArg in self.conf.dataDictionary.items():
199 if dataArg.io == 'input':
200 ignoreInputFileNames.append(dataArg.value[0])
201 # loop over all files in folder to find matching output files
202 for file in os.listdir('.'):
203 if fnmatch.fnmatch(file, pattern):
204 if file in ignoreInputFileNames:
205 msg.info('Ignoring input file: %s', file)
206 else:
207 matchedOutputFileNames.append(file)
208 return matchedOutputFileNames
209
210 # merge multiple BS files into a single BS file
211 def _mergeBSfiles(self, inputFiles, outputFile):
212 msg.info(f'Merging multiple BS files ({inputFiles}) into {outputFile}')
213 # Write the list of input files to a text file, to use it as a input for file_merging
214 mergeBSFileList = 'RAWFileMerge.list'
215 try:
216 with open(mergeBSFileList, 'w') as BSFileList:
217 for fname in inputFiles:
218 BSFileList.write(f'{fname}\n')
219 except OSError as e:
220 raise trfExceptions.TransformExecutionException(trfExit.nameToCode('TRF_OUTPUT_FILE_ERROR'),
221 f'Got an error when writing list of BS files to {mergeBSFileList}: {e}')
222
223 # The user should never need to use this directly, but check it just in case...
224 if not outputFile.endswith('._0001.data'):
225 raise trfExceptions.TransformExecutionException(trfExit.nameToCode('TRF_OUTPUT_FILE_ERROR'),
226 'Output merged BS filename must end in "._0001.data"')
227 else:
228 # We need to remove the suffix to pass the output as an argument to file_merging
229 outputFile = outputFile.split('._0001.data')[0]
230
231 mergeBSFailure = 0
232 try:
233 cmd = f'file_merging {mergeBSFileList} 0 {outputFile}'
234 msg.info('running command for merging (in original asetup env): %s', cmd)
235 mergeBSFailure = subprocess.call(cmd, shell=True)
236 msg.debug('file_merging return code %s', mergeBSFailure)
237 except OSError as e:
238 raise trfExceptions.TransformExecutionException(trfExit.nameToCode('TRF_OUTPUT_FILE_ERROR'),
239 f'Exception raised when merging BS files using file_merging: {e}')
240 if mergeBSFailure != 0:
241 msg.error('file_merging returned error (%s) no merged BS file created', mergeBSFailure)
242 return 1
243 return 0
244
245
246 # run trigbs_extractStream.py to split a stream out of the BS file
247 # renames the split file afterwards
248 def _splitBSfile(self, streamsList, allStreamsFileName, splitFileName):
249 # merge list of streams
250 outputStreams = ','.join(str(stream) for stream in streamsList)
251 msg.info('Splitting stream %s from BS file', outputStreams)
252 splitStreamFailure = 0
253 try:
254 cmd = f'trigbs_extractStream.py -s {outputStreams} {allStreamsFileName}'
255 msg.info('running command for splitting (in original asetup env): %s', cmd)
256 splitStreamFailure = subprocess.call(cmd, shell=True)
257 msg.debug('trigbs_extractStream.py splitting return code %s', splitStreamFailure)
258 except OSError as e:
259 raise trfExceptions.TransformExecutionException(trfExit.nameToCode('TRF_OUTPUT_FILE_ERROR'),
260 'Exception raised when selecting stream with trigbs_extractStream.py in file {0}: {1}'.format(allStreamsFileName, e))
261 if splitStreamFailure != 0:
262 msg.warning('trigbs_extractStream.py returned error (%s) no split BS file created', splitStreamFailure)
263 return 1
264 else:
265 # If more than one stream selected, trigbs_extractStream produces an "accepted" stream file
266 streamName = outputStreams if len(streamsList)==1 else 'accepted'
267 expectedStreamFileName = f'*_{streamName}.*.RAW._*.data'
268 # list of filenames of files matching expectedStreamFileName
269 matchedOutputFileName = self._findOutputFiles(expectedStreamFileName)
270 if(len(matchedOutputFileName)):
271 self._renamefile(matchedOutputFileName[0], splitFileName)
272 return 0
273 else:
274 msg.error('trigbs_extractStream.py did not created expected file (%s)', expectedStreamFileName)
275 return 1
276
277 # rename a created file to match the requested argument name
278 def _renamefile(self, currentFileName, newFileName):
279 msg.info('Renaming file from %s to %s', currentFileName, newFileName)
280 try:
281 os.rename(currentFileName, newFileName)
282 except OSError as e:
283 msg.error('Exception raised when renaming {0} #to {1}: {2}'.format(currentFileName, newFileName, e))
284 raise trfExceptions.TransformExecutionException(trfExit.nameToCode('TRF_OUTPUT_FILE_ERROR'),
285 'Exception raised when renaming {0} #to {1}: {2}'.format(currentFileName, newFileName, e))
286
287 def postExecute(self):
288
289 # Adding check for HLTMPPU.*Child Issue in the log file
290 # - Throws an error message if there so we catch that the child died
291 # - Also sets the return code of the mother process to mark the job as failed
292 # - Is based on trfValidation.scanLogFile
293 log = self._logFileName
294 msg.debug('Now scanning logfile {0} for HLTMPPU Child Issues'.format(log))
295 # Using the generator so that lines can be grabbed by subroutines if needed for more reporting
296
297 #Count the number of rejected events
298 rejected = 0
299 #Count the number of accepted events
300 accepted = 0
301
302 try:
303 myGen = lineByLine(log, substepName=self._substep)
304 except IOError as e:
305 msg.error('Failed to open transform logfile {0}: {1:s}'.format(log, e))
306 for line, lineCounter in myGen:
307 # Check to see if any of the hlt children had an issue
308 if 'Child Issue' in line:
309 try:
310 signal = int((re.search('signal ([0-9]*)', line)).group(1))
311 except AttributeError:
312 # signal not found in message, so return 1 to highlight failure
313 signal = 1
314 msg.error('Detected issue with HLTChild, setting mother return code to %s', signal)
315 self._rc = signal
316
317 # Merge child log files into parent log file
318 # is needed to make sure all child log files are scanned
319 # files are found by searching whole folder rather than relying on nprocs being defined
320 try:
321 # open original log file (log.BSRDOtoRAW) to merge child files into
322 with open(self._logFileName, 'a') as merged_file:
323 for file in os.listdir('.'):
324 # expected child log files should be of the format athenaHLT:XX.out and .err
325 if fnmatch.fnmatch(file, 'athenaHLT:*'):
326 msg.info('Merging child log file (%s) into %s', file, self._logFileName)
327 with open(file) as log_file:
328 # write header infomation ### Output from athenaHLT:XX.out/err ###
329 merged_file.write('### Output from {} ###\n'.format(file))
330 # write out file line by line
331 for line in log_file:
332 merged_file.write(line)
333 # Check for rejected events in log file
334 if 'rejected:' in line and int(line[14]) != 0:
335 #Add the number of rejected events
336 rejected += int(line[14:])
337 # Check for accepted events in log file
338 if 'accepted:' in line and int(line[14]) != 0:
339 #Add the number of accepted events
340 accepted += int(line[14:])
341 if re.search('DFDcmEmuSession.* Communication error', line) or re.search('DFDcmEmuSession.* No new event provided within the timeout limit', line):
342 msg.error('Caught DFDcmEmuSession error, aborting job')
343 self._rc = 1
344
345 if "HIST_DEBUGSTREAMMON" in self.conf.dataDictionary:
346 # Add the HLT_accepted_events and HLT_rejected_events histograms to the output file
347 dbgStream.getHltDecision(accepted, rejected, self.conf.argdict["outputHIST_DEBUGSTREAMMONFile"].value[0])
348
349 except OSError as e:
350 raise trfExceptions.TransformExecutionException(trfExit.nameToCode('TRF_OUTPUT_FILE_ERROR'),
351 'Exception raised when merging log files into {0}: {1}'.format(self._logFileName, e))
352
353 msg.info("Check for expert-monitoring.root file")
354 # the BS-BS step generates the files:
355 # - expert-monitoring.root (from mother process)
356 # - athenaHLT_workers/*/expert-monitoring.root (from child processes)
357 # to save on panda it needs to be renamed via the outputHIST_HLTMONFile argument
358 expectedFileName = 'expert-monitoring.root'
359
360 # first check if trigger step actually completed
361 if self._rc != 0:
362 msg.info('HLT step failed (with status %s) so skip HIST_HLTMON filename check', self._rc)
363 # next check argument is in dictionary as a requested output
364 elif 'outputHIST_HLTMONFile' in self.conf.argdict:
365
366 # rename the mother file
367 expectedMotherFileName = 'expert-monitoring-mother.root'
368 if(os.path.isfile(expectedFileName)):
369 msg.info('Renaming %s to %s', expectedFileName, expectedMotherFileName)
370 try:
371 os.rename(expectedFileName, expectedMotherFileName)
372 except OSError as e:
373 raise trfExceptions.TransformExecutionException(trfExit.nameToCode('TRF_OUTPUT_FILE_ERROR'),
374 'Exception raised when renaming {0} to {1}: {2}'.format(expectedFileName, expectedMotherFileName, e))
375 else:
376 msg.error('HLTMON argument defined but mother %s not created', expectedFileName)
377
378 # merge worker files
379 expectedWorkerFileName = 'athenaHLT_workers/athenaHLT-01/' + expectedFileName
380 if(os.path.isfile(expectedWorkerFileName) and os.path.isfile(expectedMotherFileName)):
381 msg.info('Merging worker and mother %s files to %s', expectedFileName, self.conf.argdict['outputHIST_HLTMONFile'].value[0])
382 try:
383 # have checked that at least one worker file exists
384 cmd = 'hadd ' + self.conf.argdict['outputHIST_HLTMONFile'].value[0] + ' athenaHLT_workers/*/expert-monitoring.root expert-monitoring-mother.root'
385 subprocess.call(cmd, shell=True)
386 except OSError as e:
387 raise trfExceptions.TransformExecutionException(trfExit.nameToCode('TRF_OUTPUT_FILE_ERROR'),
388 'Exception raised when merging worker and mother {0} files to {1}: {2}'.format(expectedFileName, self.conf.argdict['outputHIST_HLTMONFile'].value[0], e))
389 else:
390 msg.error('HLTMON argument defined %s but worker %s not created', self.conf.argdict['outputHIST_HLTMONFile'].value[0], expectedFileName)
391
392 else:
393 msg.info('HLTMON argument not defined so skip %s check', expectedFileName)
394
395 msg.info("Search for created BS files, and rename if single file found")
396 # The following is needed to handle the BS file being written with a different name (or names)
397 # base is from either the tmp value created by the transform or the value entered by the user
398
399 argInDict = {}
400 if self._rc != 0:
401 msg.error('HLT step failed (with status %s) so skip BS filename check', self._rc)
402 elif 'BS' in self.conf.dataDictionary or 'DRAW_TRIGCOST' in self.conf.dataDictionary or 'HIST_DEBUGSTREAMMON' in self.conf.dataDictionary:
403 # list of filenames of files matching expectedOutputFileName
404 matchedOutputFileNames = self._findOutputFiles(self.expectedOutputFileName)
405
406
407 # check there are file matches and rename appropriately
408 if len(matchedOutputFileNames) == 0:
409 msg.error('No BS files created with expected name: %s - please check for earlier failures', self.expectedOutputFileName)
410 raise trfExceptions.TransformExecutionException(trfExit.nameToCode('TRF_OUTPUT_FILE_ERROR'),
411 f'No BS files created with expected name: {self.expectedOutputFileName} - please check for earlier failures')
412 else:
413 # if only one BS file was created
414 if len(matchedOutputFileNames) == 1:
415 msg.info('Single BS file found: will split (if requested) and rename the file')
416 BSFile = matchedOutputFileNames[0]
417
418 # if more than one file BS was created, then merge them
419 else:
420 msg.info('Multiple BS files found. A single BS file is required by the next transform steps, so they will be merged. Will split the merged file (if requested) and rename the file')
421 mergedBSFile = matchedOutputFileNames[0].split('._0001.data')[0] + '.mrg._0001.data' # a specific format is required to run file_merging
422
423 mergeFailed = self._mergeBSfiles(matchedOutputFileNames, mergedBSFile)
424 if(mergeFailed):
425 raise trfExceptions.TransformExecutionException(trfExit.nameToCode('TRF_OUTPUT_FILE_ERROR'),
426 'Did not produce a merged BS file with file_merging')
427
428 BSFile = 'tmp.BS.mrg'
429 msg.info(f'Renaming temporary merged BS file to {BSFile}')
430 self._renamefile(mergedBSFile, BSFile)
431
432 # First check if we want to produce the COST DRAW output
433 if 'DRAW_TRIGCOST' in self.conf.dataDictionary:
434 splitFailed = self._splitBSfile(['CostMonitoring'], BSFile, self.conf.dataDictionary['DRAW_TRIGCOST'].value[0])
435 if(splitFailed):
436 raise trfExceptions.TransformExecutionException(trfExit.nameToCode('TRF_OUTPUT_FILE_ERROR'),
437 'Did not produce any BS file when selecting CostMonitoring stream with trigbs_extractStream.py in file')
438
439 # Run debug step for all streams
440 if "HIST_DEBUGSTREAMMON" in self.conf.dataDictionary:
441 self._postExecuteDebug(BSFile)
442
443 # Rename BS file if requested
444 if 'BS' in self.conf.dataDictionary:
445 argInDict = self.conf.dataDictionary['BS']
446 # If a stream (not All) is selected, then slim the orignal (many stream) BS output to the particular stream
447 if 'streamSelection' in self.conf.argdict and self.conf.argdict['streamSelection'].value[0] != "All":
448 splitEmpty = self._splitBSfile(self.conf.argdict['streamSelection'].value, BSFile, argInDict.value[0])
449 if(splitEmpty):
450 msg.info('Did not produce any BS file when selecting stream with trigbs_extractStream.py in file')
451 #If splitEmpty==1, the chosen streams contained no events
452 #then run the command to produce an empty BS file and rename it to RAW.pool.root
453 #this stops non-zero exit code for rejected events
454 cmd_splitFailed = 'trigbs_failedStreamSelection.py ' + BSFile
455 msg.info('running command for creating empty file: %s', cmd_splitFailed)
456 subprocess.call(cmd_splitFailed, shell=True)
457 #Rename the empty file to "RAW.pool.root" to prevent failure
458 #expected filename will be of form: T0debug.runnumber.unknown_debug.unknown.RAW._lb0000._TRF._0001.data
459 runnumber = eformat.EventStorage.pickDataReader(BSFile).runNumber()
460 expectedOutputFileName = 'T0debug.00'+str(runnumber)+'.unknown_debug.unknown.RAW._lb0000._TRF._0001.data'
461 #rename the file to RAW.pool.root, this file will contain 0 events
462 self._renamefile(expectedOutputFileName, argInDict.value[0])
463 else:
464 msg.info('Stream "All" requested, so not splitting BS file')
465 self._renamefile(BSFile, argInDict.value[0])
466 else:
467 msg.info('BS output filetype not defined so skip renaming BS')
468 else:
469 msg.info('BS, DRAW_TRIGCOST or HIST_DEBUGSTREAMMON output filetypes not defined so skip BS post processing')
470
471 msg.info('Now run athenaExecutor:postExecute')
472 super(trigRecoExecutor, self).postExecute()
473
474 if "HIST_DEBUGSTREAMMON" in self.conf.dataDictionary:
475 # Do debug stream postRun step for BS file that contains events after the streamSelection
476 fileNameDbg = self.conf.argdict["outputHIST_DEBUGSTREAMMONFile"].value
477 dbgStream.dbgPostRun(argInDict.value[0], fileNameDbg[0], self.conf.argdict, isSplitStream=True)
478
479
480 def _postExecuteDebug(self, outputBSFile):
481 # Run postRun step debug stream analysis if output BS file and output histogram are set
482 msg.info("debug stream analysis in postExecute")
483
484 # Set file name for debug stream analysis output
485 fileNameDbg = self.conf.argdict["outputHIST_DEBUGSTREAMMONFile"].value
486 msg.info('outputHIST_DEBUGSTREAMMONFile argument is {0}'.format(fileNameDbg))
487
488 if(os.path.isfile(fileNameDbg[0])):
489 # Keep filename if not defined
490 msg.info('Will use file created in PreRun step {0}'.format(fileNameDbg))
491 else:
492 msg.info('No file created in PreRun step {0}'.format(fileNameDbg))
493
494 # Do debug stream postRun step
495 dbgStream.dbgPostRun(outputBSFile, fileNameDbg[0], self.conf.argdict)
496
if(pathvar)
Class holding the update to an environment that will be passed on to an executor.
Definition trfEnv.py:15
Base class for execution exceptions.
_splitBSfile(self, streamsList, allStreamsFileName, splitFileName)
_renamefile(self, currentFileName, newFileName)
_postExecuteDebug(self, outputBSFile)
preExecute(self, input=set(), output=set())
_mergeBSfiles(self, inputFiles, outputFile)
STL class.
std::vector< std::string > split(const std::string &s, const std::string &t=":")
Definition hcg.cxx:179
Transform execution functions.
Module for transform exit codes.
Transform utility functions.