Herwig7Utils.py

Go to the documentation of this file.

# Copyright (C) 2002-2020 CERN for the benefit of the ATLAS collaboration
 
 
 
import math, os, subprocess
 
from AthenaCommon import Logging
athMsgLog = Logging.logging.getLogger('Herwig7Utils')
 
integration_grids_precision_threshold = 0.0005 #  if integration xsec below a warning is issue in debug mode and an erroro in production mode
 
 
 
class ConfigurationCommands:
 
  def __init__(self):
    self.commands = ""
    self.locked = False
  
  def lock(self):
    self.locked = True
  
  def add(self, commands):
    if self.locked is False:
      self.commands += commands
    else:
      raise RuntimeError("The commands associated to this configuration object can't be modified anymore because the Herwig7 infile has already been written to disk. Therefore, any subsequent additional modifications can't be adopted in the Herwig7 run.")
    return(self)
  
  __iadd__ = add
 
 
 
class ProcessHandler:
 
  def __init__(self, process_list, logger):
    self.processes = process_list
    self.n_initial = len(process_list)
    self.logger = logger
 
  def success(self):
    result = True
    while len(self.processes) > 0:
      for ID, process in enumerate(self.processes):
        returncode = process.poll()
        if returncode is not None:
 
          
          self.processes.remove(process)
          if returncode == 0:
            athMsgLog.info(ansi_format_info('Process #{} finished sucessfully, {}/{} still running'.format(process.ID, len(self.processes), self.n_initial)))
          elif returncode > 0:
            athMsgLog.error(ansi_format_error("Process #{} finished with error code {} (please check logfile '{}'), {}/{} still running".format(process.ID, returncode, process.logfile_title, len(self.processes), self.n_initial)))
            result = False
          elif returncode < 0:
            athMsgLog.error(ansi_format_error("Process #{} was terminated by signal {} (please check logfile '{}'), {}/{} still running".format(process.ID, -returncode, process.logfile_title, len(self.processes), self.n_initial)))
            result = False
 
          
          athMsgLog.info("Content of integration log file '%s':", process.logfile_title)
          athMsgLog.info("")
          with open(process.logfile_title, 'r') as logfile:
            for line in logfile:
              athMsgLog.info('  %s', line.rstrip('\n'))
          athMsgLog.info("")
          # self.logger.info('================================================================================')
          # self.logger.info("End of integration log file '{}'".format(process.logfile_title))
          # self.logger.info('================================================================================')
 
    return(result)
 
 
 
class Process:
 
  def __init__(self, ID, command, logfile_title):
 
    athMsgLog.info(ansi_format_info("Starting subprocess #{} with command '{}' and logfile '{}'".format(ID, ' '.join(command), logfile_title)))
    self.ID = ID
    self.logfile_title = logfile_title
    self.logfile = open(logfile_title, 'w')
    self.process = subprocess.Popen(command, stdout=self.logfile, stderr=self.logfile)
 
  def poll(self):
    return(self.process.poll())
 
 
 
def get_cross_section(run_name, integration_jobs=1):
 
  athMsgLog.info("Calculating cross section after integration")
  logfiles = [run_name+'.integrate'+str(integration_job)+'.log' for integration_job in range(integration_jobs)]
 
  xsec = 0.0
  err  = 0.0
 
  for logfile in logfiles:
 
    athMsgLog.info("- %s", logfile)
 
    
    with open(logfile, 'r') as log: data = log.read().strip()
 
    with open(logfile, 'r') as log:
 
      
      for line in log:
        if 'Integrate ' in line:
          n_subprocs = int(line.split('of')[1].replace(':',''))
          athMsgLog.info("  found %s subprocesses", n_subprocs)
          break
 
    data = data.split("Integrate ")[1:]
  
    for s, subproc in enumerate(data, start=1):
      _xsec = 0.0
      _err  = 0.0
      for line in subproc.split("\n"):
        if 'integrated ( ' in line:
          _xsec = float(line.split()[2])
          _err = float(line.split()[4])
      athMsgLog.info("  - subprocess %s: xsec = %s +/- %s nb", s, _xsec, _err)
      xsec += _xsec
      err  += _err*_err
 
  err = math.sqrt(err)
 
  if err / xsec > integration_grids_precision_threshold:
    threshold = '{}%'.format(integration_grids_precision_threshold*100.0)
    if run_name == "Herwig_DEBUG":
       athMsgLog.warn(ansi_format_warning('! The integration grids only have a low precision (worse than {}): xsec = {} +/- {} nb (accuracy: {:.3f}%)'.format(threshold, xsec, err, err/xsec*100.0)))
    else:
       athMsgLog.error(ansi_format_error('! The integration grids only have a low precision (worse than {}): xsec = {} +/- {} nb (accuracy: {:.3f}%)'.format(threshold, xsec, err, err/xsec*100.0)))
    athMsgLog.warn(ansi_format_warning('! In order to speed up the event generation you should consider improving the statistics of the integration / phase space sampling stage (see the sampler_commands() function).'))
  else:
    athMsgLog.info(ansi_format_info('After integration the estimated cross section was found to be: xsec = {} +/- {} nb (accuracy: {:.3f}%)'.format(xsec, err, err/xsec*100.0)))
 
  return(xsec, err)
 
 
 
def get_repeated_pattern(pattern, repetitions):
 
  return(pattern.join(['' for i in range(repetitions+1)]))
 
 
 
def get_size(path):
 
  size = 0
  
  if os.path.isfile(path):
    return(os.path.getsize(path))
 
  elif os.path.isdir(path):
    for dirpath, dirs, files in os.walk(path):
      for file in files:
        file_path = os.path.join(dirpath, file)
        # if not os.path.islink(file_path): # exclude symlinks because MadGraph has lots of broken symlinks
        if os.path.isfile(file_path):
          size += os.path.getsize(file_path)
 
  return(size)
 
 
 
def humanize_bytes(bytes, precision=2):
  # shamelessly stolen from
  # http://code.activestate.com/recipes/577081-humanized-representation-of-a-number-of-bytes/
 
  abbrevs = (
      (1<<50, 'PiB'),
      (1<<40, 'TiB'),
      (1<<30, 'GiB'),
      (1<<20, 'MiB'),
      (1<<10, 'KiB'),
      (1, 'byte(s)')
  )
  if bytes == 1:
    return '1 byte'
  for factor, suffix in abbrevs:
    if bytes >= factor:
      break
  return '%.*f %s' % (precision, bytes / factor, suffix)
 
 
 
def ansi_format(text, colour='None', background_colour='None',
           bold=False, underline=False, italics=False, marked=False, strikethrough=False):
 
  format_none          = '\033[0m'
  format_bold          = '\033[1m'
  format_underline     = '\033[4m'
  format_italics       = '\033[3m'
  format_marked        = '\033[7m'
  format_strikethrough = '\033[9m'
 
  colours = {
    'None': '',
    'Black':     '\033[30m',
    'Grey':      '\033[90m',
    'Red':       '\033[91m', 'DarkRed':   '\033[31m',
    'Green':     '\033[92m', 'DarkGreen': '\033[32m',
    'Yellow':    '\033[93m', 'Orange':    '\033[33m',
    'Blue':      '\033[94m', 'DarkBlue':  '\033[34m',
    'Pink':      '\033[95m', 'DarkPink':  '\033[35m',
    'Cyan':      '\033[96m', 'DarkCyan':  '\033[36m',
    'White':     '\033[97m', 'LightGrey': '\033[37m',
  }
 
  background_colours = {
    'None': '',
    'Black':  '\033[40m',
    'Red':    '\033[41m',
    'Green':  '\033[42m',
    'Orange': '\033[43m',
    'Blue':   '\033[44m',
    'Punk':   '\033[45m',
    'Cyan':   '\033[46m',
    'Grey':   '\033[47m',
  }
 
  if colour in colours:
    if background_colour in background_colours:
      return (format_none
        + colours[colour]
        + background_colours[background_colour]
        + (format_bold          if bold          else '')
        + (format_underline     if underline     else '')
        + (format_italics       if italics       else '')
        + (format_marked        if marked        else '')
        + (format_strikethrough if strikethrough else '')
        + text + format_none)
    else:
      raise Exception("Could not find background colour '{}'.".format(background_colour))
  else:
    raise Exception("Could not find colour '{}'.".format(colour))
 
 
def ansi_format_ok(text):
  return (ansi_format(text, "Green"))
 
 
def ansi_format_info(text):
  return (ansi_format(text, "Blue"))
 
 
def ansi_format_warning(text):
  return (ansi_format(text, "Yellow"))
 
 
def ansi_format_error(text):
  return (ansi_format(text, "Red"))