Tools/PyUtils/python/Helpers.py

Go to the documentation of this file.

# Copyright (C) 2002-2026 CERN for the benefit of the ATLAS collaboration
 
# @author: Sebastien Binet <binet@cern.ch>
# @date:   March 2007
#
#
 
import re
import os
import sys
from functools import cache
from tempfile import NamedTemporaryFile
# !!!
# Do not add any Athena dependencies at the module-level as some of the code here
# is used during the build (see athena!62739).
# !!!
 
 
def ROOTSetup(batch=True):
   """
   Do basic ROOT setup. This should be called before any other import of ROOT.
   """
   from AthenaCommon.Logging import log
   log.debug('Setting ROOT batch mode to %s', batch)
 
   if batch is True:
      os.environ['ROOT_BATCH'] = '1'
   else:
      os.environ.pop('ROOT_BATCH', None)
 
 
class ShutUp(object):
    """
    A little helper class to keep ROOT silent...
    """
    DefaultFilter = [
        re.compile("Warning in <TClass::TClass>: no dictionary for class."),
        re.compile("Warning in <TEnvRec::ChangeValue>: duplicate entry."),
        re.compile("Error in <TStreamerInfo::BuildOld>:."),
        ]
    def __init__(self, filters = DefaultFilter):
        self._dummy = False # if dummy, we don't really shut-up ROOT...
        if os.environ.get('PYUTILS_SHUTUP_DEBUG', '0') == '1':
            self._dummy = True
            
        self.save_err  = open( '/dev/null', 'w' )
        self.save_out  = open( '/dev/null', 'w' )
        try:
            self.quiet_err = NamedTemporaryFile( suffix = ".msg.log" )
            self.quiet_out = NamedTemporaryFile( suffix = ".msg.log" )
        except OSError:
            # load problem ?
            # retry a bit later... (in 2 seconds)
            import time
            time.sleep(2) #
            try:
                self.quiet_err = NamedTemporaryFile( suffix = ".msg.log" )
                self.quiet_out = NamedTemporaryFile( suffix = ".msg.log" )
            except OSError:
                # then (implicitly) fallback on sys.stderr
                self._dummy = True
 
        self.filters = filters
 
        if not self._dummy:
            os.dup2( sys.stderr.fileno(), self.save_err.fileno() )
            os.dup2( sys.stdout.fileno(), self.save_out.fileno() )
        return
 
    def __del__ (self):
       self.save_err.close()
       self.save_out.close()
       return
    
    def mute(self):
        if not self._dummy:
            os.dup2( self.quiet_err.fileno(), sys.stderr.fileno() )
            os.dup2( self.quiet_out.fileno(), sys.stdout.fileno() )
        return
    
    def unMute(self):
        if not self._dummy:
            os.dup2( self.save_err.fileno(), sys.stderr.fileno() )
            os.dup2( self.save_out.fileno(), sys.stdout.fileno() )
            self.__filterRootMessages(self.quiet_err)
            self.quiet_err.seek(0)
            self.__filterRootMessages(self.quiet_out)
            self.quiet_out.seek(0)
        return
 
    def __filterRootMessages(self, fd):
        fd.seek(0)
        for l in fd.readlines():
            l = l.decode()
            printOut = True
            for filter in self.filters:
                if re.match(filter, l):
                    printOut = False
            if printOut:
                print ("PyRoot:",l.replace("\n",""))
            pass
        return
 
    # context-manager protocol
    def __enter__(self):
        return self.mute()
    def __exit__(self,exc_type, exc_val, exc_tb):
        return self.unMute()
 
 
@cache
def release_metadata():
   """Returns information about the current release based on ReleaseData"""
 
   import configparser
   d = {
      'project name': '?',
      'release': '?',
      'base release': '?',
      'nightly release': '?',
      'nightly name': '?',
      'date': '?',
      'platform': '?',
      }
 
   for cmake_path in os.environ['CMAKE_PREFIX_PATH'].split(os.pathsep):
      release_data = os.path.join(cmake_path, 'ReleaseData')
      if os.path.exists(release_data):
         d1=d
         cfg = configparser.ConfigParser()
         try:
            cfg.read( release_data )
            if cfg.has_section( 'release_metadata' ):
               d1.update( dict( cfg.items( 'release_metadata' ) ) )
               d1['platform'] = os.getenv( '%s_PLATFORM' % d1['project name'],
                                           '?' )
               release = d1['release'].split('.')
               base_release = d1['base release'].split('.')
               if len(release)>=3 or len(base_release)>=3:
                  return d1
         except Exception:
            pass
   return d