File indexing completed on 2023-03-17 10:40:23
0001 from __future__ import print_function
0002 from __future__ import absolute_import
0003
0004 import configparser as ConfigParser
0005 import os
0006 import re
0007 import copy
0008 import collections
0009 from .TkAlExceptions import AllInOneError
0010 from future.utils import PY3
0011
0012 if PY3:
0013 unicode = str
0014
0015 class AdaptedDict(collections.OrderedDict):
0016 """
0017 Dictionary which handles updates of values for already existing keys
0018 in a modified way.
0019 adapteddict[key] returns a list of all values associated with key
0020 This dictionary is used in the class `BetterConfigParser` instead of the
0021 default `dict_type` of the `ConfigParser` class.
0022 """
0023
0024 def __init__(self, *args, **kwargs):
0025 self.validationslist = []
0026 collections.OrderedDict.__init__(self, *args, **kwargs)
0027
0028 def __setitem__(self, key, value, dict_setitem=collections.OrderedDict.__setitem__):
0029 """
0030 od.__setitem__(i, y) <==> od[i]=y
0031 Updating an existing key appends the new value to the old value
0032 instead of replacing it.
0033
0034 Arguments:
0035 - `key`: key part of the key-value pair
0036 - `value`: value part of the key-value pair
0037 - `dict_item`: method which is used for finally setting the item
0038 """
0039
0040 if key != "__name__" and "__name__" in self and self["__name__"]=="validation":
0041 if isinstance(value, (str, unicode)):
0042 for index, item in enumerate(self.validationslist[:]):
0043 if item == (key, value.split("\n")):
0044 self.validationslist[index] = (key, value)
0045 return
0046 self.validationslist.append((key, value))
0047 else:
0048 dict_setitem(self, key, value)
0049
0050 def __getitem__(self, key):
0051 if key != "__name__" and "__name__" in self and self["__name__"]=="validation":
0052 return [validation[1] for validation in self.validationslist if validation[0] == key]
0053 else:
0054 return collections.OrderedDict.__getitem__(self, key)
0055
0056 def items(self):
0057 if "__name__" in self and self["__name__"]=="validation":
0058 return self.validationslist
0059 else:
0060 return collections.OrderedDict.items(self)
0061
0062 class BetterConfigParser(ConfigParser.ConfigParser):
0063 def __init__(self):
0064 ConfigParser.ConfigParser.__init__(self,dict_type=AdaptedDict)
0065 self._optcre = self.OPTCRE_VALIDATION
0066
0067 def optionxform(self, optionstr):
0068 return optionstr
0069
0070 def exists( self, section, option):
0071 try:
0072 items = self.items(section)
0073 except ConfigParser.NoSectionError:
0074 return False
0075 for item in items:
0076 if item[0] == option:
0077 return True
0078 return False
0079
0080 def __updateDict( self, dictionary, section ):
0081 result = dictionary
0082 try:
0083 for option in self.options( section ):
0084 result[option] = self.get( section, option )
0085 if "local"+section.title() in self.sections():
0086 for option in self.options( "local"+section.title() ):
0087 result[option] = self.get( "local"+section.title(),
0088 option )
0089 except ConfigParser.NoSectionError as section:
0090 msg = ("%s in configuration files. This section is mandatory."
0091 %(str(section).replace(":", "", 1)))
0092 raise AllInOneError(msg)
0093 return result
0094
0095 def getResultingSection( self, section, defaultDict = {}, demandPars = [] ):
0096 result = copy.deepcopy(defaultDict)
0097 for option in demandPars:
0098 try:
0099 result[option] = self.get( section, option )
0100 except ConfigParser.NoOptionError as globalSectionError:
0101 globalSection = str( globalSectionError ).split( "'" )[-2]
0102 splittedSectionName = section.split( ":" )
0103 if len( splittedSectionName ) > 1:
0104 localSection = ("local"+section.split( ":" )[0].title()+":"
0105 +section.split(":")[1])
0106 else:
0107 localSection = ("local"+section.split( ":" )[0].title())
0108 if self.has_section( localSection ):
0109 try:
0110 result[option] = self.get( localSection, option )
0111 except ConfigParser.NoOptionError as option:
0112 msg = ("%s. This option is mandatory."
0113 %(str(option).replace(":", "", 1).replace(
0114 "section",
0115 "section '"+globalSection+"' or", 1)))
0116 raise AllInOneError(msg)
0117 else:
0118 msg = ("%s. This option is mandatory."
0119 %(str(globalSectionError).replace(":", "", 1)))
0120 raise AllInOneError(msg)
0121 try:
0122 result = self.__updateDict( result, section )
0123 except AllInOneError:
0124 if demandPars:
0125 raise
0126 return result
0127
0128 def getAlignments( self ):
0129 alignments = []
0130 for section in self.sections():
0131 if "alignment:" in section:
0132 alignments.append( Alignment( section.split( "alignment:" )[1],
0133 self ) )
0134 names_after_cleaning = [alignment.name for alignment in alignments]
0135 duplicates = [name
0136 for name, count
0137 in collections.Counter(names_after_cleaning).items()
0138 if count > 1]
0139 if len(duplicates) > 0:
0140 msg = "Duplicate alignment names after removing invalid characters: "
0141 msg += ", ".join(duplicates) +"\n"
0142 msg += "Please rename the alignments to avoid name clashes."
0143 raise AllInOneError(msg)
0144 return alignments
0145
0146 def getCompares( self ):
0147 compares = {}
0148 for section in self.sections():
0149 if "compare:" in section:
0150 self.checkInput(section,
0151 knownSimpleOptions = ["levels", "dbOutput","moduleList","modulesToPlot","useDefaultRange","plotOnlyGlobal","plotPng","makeProfilePlots",
0152 "dx_min","dx_max","dy_min","dy_max","dz_min","dz_max","dr_min","dr_max","rdphi_min","rdphi_max",
0153 "dalpha_min","dalpha_max","dbeta_min","dbeta_max","dgamma_min","dgamma_max",
0154 "jobmode", "3DSubdetector1", "3Dubdetector2", "3DTranslationalScaleFactor", "jobid", "multiIOV"])
0155 levels = self.get( section, "levels" )
0156 dbOutput = self.get( section, "dbOutput" )
0157 compares[section.split(":")[1]] = ( levels, dbOutput )
0158 return compares
0159
0160 def getGeneral( self ):
0161 defaults = {
0162 "jobmode":"interactive",
0163 "datadir":os.getcwd(),
0164 "logdir":os.getcwd(),
0165 }
0166 mandatories = ["eosdir",]
0167 self.checkInput("general", knownSimpleOptions = list(defaults.keys()) + mandatories )
0168 general = self.getResultingSection( "general", defaultDict = defaults, demandPars = mandatories )
0169 internal_section = "internals"
0170 if not self.has_section(internal_section):
0171 self.add_section(internal_section)
0172 if not self.has_option(internal_section, "workdir"):
0173 self.set(internal_section, "workdir", "/tmp/$USER")
0174 if not self.has_option(internal_section, "scriptsdir"):
0175 self.set(internal_section, "scriptsdir", "")
0176
0177
0178 general["workdir"] = self.get(internal_section, "workdir")
0179 general["scriptsdir"] = self.get(internal_section, "scriptsdir")
0180 for folder in "workdir", "datadir", "logdir", "eosdir":
0181 general[folder] = os.path.expandvars(general[folder])
0182
0183 return general
0184
0185 def checkInput(self, section, knownSimpleOptions=[], knownKeywords=[],
0186 ignoreOptions=[]):
0187 """
0188 Method which checks, if the given options in `section` are in the
0189 list of `knownSimpleOptions` or match an item of `knownKeywords`.
0190 This is basically a check for typos and wrong parameters.
0191
0192 Arguments:
0193 - `section`: Section of a configuration file
0194 - `knownSimpleOptions`: List of allowed simple options in `section`.
0195 - `knownKeywords`: List of allowed keywords in `section`.
0196 """
0197
0198 try:
0199 for option in self.options( section ):
0200 if option in knownSimpleOptions:
0201 continue
0202 elif option.split()[0] in knownKeywords:
0203 continue
0204 elif option in ignoreOptions:
0205 print ("Ignoring option '%s' in section '[%s]'."
0206 %(option, section))
0207 else:
0208 msg = ("Invalid or unknown parameter '%s' in section '%s'!"
0209 %(option, section))
0210 raise AllInOneError(msg)
0211 except ConfigParser.NoSectionError:
0212 pass
0213
0214 def set(self, section, option, value=None):
0215 try:
0216 ConfigParser.ConfigParser.set(self, section, option, value)
0217 except ConfigParser.NoSectionError:
0218 self.add_section(section)
0219 ConfigParser.ConfigParser.set(self, section, option, value)
0220
0221 def items(self, section, raw=False, vars=None):
0222 if section == "validation":
0223 if raw or vars:
0224 raise NotImplementedError("'raw' and 'vars' do not work for betterConfigParser.items()!")
0225 items = self._sections["validation"].items()
0226 return items
0227 else:
0228 return ConfigParser.ConfigParser.items(self, section, raw, vars)
0229
0230 def write(self, fp):
0231 """Write an .ini-format representation of the configuration state."""
0232 for section in self._sections:
0233 fp.write("[%s]\n" % section)
0234 for (key, value) in self._sections[section].items():
0235 if key == "__name__" or not isinstance(value, (str, unicode)):
0236 continue
0237 if value is not None:
0238 key = " = ".join((key, str(value).replace('\n', '\n\t')))
0239 fp.write("%s\n" % (key))
0240 fp.write("\n")
0241
0242
0243
0244
0245
0246 OPTCRE_VALIDATION = re.compile(
0247 r'(?P<option>'
0248 r'(?P<preexisting>preexisting)?'
0249 r'[^:=\s][^:=]*)'
0250 r'\s*(?(preexisting)|'
0251 r'(?P<vi>[:=])\s*'
0252
0253
0254
0255 r'(?P<value>.*))$'
0256 )
